Whenever a piece of software obtains outside input, it's a good idea to be sure the input is valid for what you're expecting. The web world uses client-side JavaScript as well as various server-side frameworks to validate user input, and validation of batch parameters is no different. Fortunately, Spring has made it very easy to validate job parameters. To do so, you just need to implement the org.springframework.batch.core.JobParametersValidator interface and configure your implementation in your job. Listing 4-11 shows an example of a job parameter validator in Spring Batch.

package com.apress.springbatch.chapter4;

import java.util.Map;
import org.springframework.batch.core.*;
import org.apache.commons.lang.StringUtils;

public class ParameterValidator implements JobParametersValidator{

    public void validate(JobParameters params) throws
        JobParametersInvalidException {
        String name = params.getString("name");
        if(!StringUtils.isAlpha(name)) {
            throw new
                JobParametersInvalidException("Name is not alphabetic");
        }
    }
}

As you can see, the method of consequence is the validate method. Because this method is void, the validation is considered passing as long as a JobParametersInvalidException isn't thrown. In this example, if you pass the name 4566, the exception is thrown and the job completes with a status of COMPLETED. This is important to note. Just because the parameters you passed in weren't valid doesn't mean the job didn't complete correctly. In the case where invalid parameters are passed, the job is marked as COMPLETED because it did all valid processing for the input it received. And when you think about this, it makes sense. A JobInstance is identified by the job name and the parameters passed into the job. If you pass in invalid parameters, you don't want to repeat that, so it's ok to declare the job completed.

In addition to implementing your own custom parameter validator as you did earlier, Spring Batch offers a validator to confirm that all the required parameters have been passed: org.springframework.batch.core.job.DefaultJobParametersValidator. To use it, you configure it the same way you would your custom validator. DefaultJobParametersValidator has two optional dependencies: requiredKeys and optionalKeys. Both are String arrays that take in a list of parameter names that are either required or are the only optional parameters allowed. Listing 4-12 shows two configurations for DefaultJobParametersValidator as well as how to add it to your job.

<beans:bean id="requiredParamValidator"
    class="org.springframework.batch.core.job.DefaultJobParametersValidator">
    <beans:property name="requiredKeys" value="batch.name,batch.runDate"/>
</beans:bean>

<beans:bean id="optionalParamValidator"
    class="org.springframework.batch.core.job.DefaultJobParametersValidator">
    <beans:property name="requiredKeys" value="batch.name,batch.runDate"/>
    <beans:property name="optionalKeys" value="batch.address"/>
</beans:bean>

<job id="parameterValidatorJob">
    ...
    <validator ref="requiredParamValidator"/>
</job>

If you use requiredParamValidator, your job throws an exception if you don't pass the parameters batch.name and batch.runDate. You're allowed to pass more parameters in if required, but those two can't be null. On the other hand, if you use optionalParamValidator, the job once again throws an exception if batch.name and batch.runDate aren't passed to the job, but it also throws an exception if any parameters in addition to batch.address are passed. The difference between the two validators is that the first one can accept any parameters in addition to the required ones. The second one can only accept the three specified. In either case, if the invalid scenario occurs, a JobParametersInvalidException is thrown and the job is marked as completed as previously discussed.

Up to now, you've been running under the limitation that a job can only be run once with a given set of parameters. If you've been following along with the examples, you've probably hit what happens if you attempt to run the same job twice with the same parameters as shown in Listing 4-5. However, there is a small loophole: using JobParametersIncrementer.

org.springframework.batch.core.JobParametersIncrementer is an interface that Spring Batch provides to allow you to uniquely generate parameters for a given job. You can add a timestamp to each run. You may have some other business logic that requires a parameter to be incremented with each run. The framework provides a single implementation of the interface, which increments a single long parameter with the default name run.id.

Listing 4-13 shows how to configure a JobParametersIncrementer for your job by adding the reference to the job.

<beans:bean id="idIncrementer"
    class="org.springframework.batch.core.launch.support.RunIdIncrementer"/>

<job id="baseJob" incrementer="idIncrementer">
    ...
</job>

Once you've configured JobParametersIncrementer (the framework provides org.springframework.batch.core.launch.support.RunIdIncrementer in this case), there are two more things you need to do to make this work. First you need to add the configuration for a JobExplorer implementation. Chapter 5 goes into detail about what JobExplorer is and how to use it. For now, just know that Spring Batch needs it to increment parameters. Listing 4-14 shows the configuration, but it's already configured in the launch-context.xml that is included in the zip file distribution.

<bean id="jobExplorer"
class="org.springframework.batch.core.explore.support.JobExplorerFactoryBean">
    <property name="dataSource" ref="dataSource"/>
</bean>

The last piece of the puzzle to use a JobParametersIncrementer affects how you call your job. When you want to increment a parameter, you need to add the parameter –next to the command when you call your job. This tells Spring Batch to use the incrementer as required.

Now when you run your job with the command in Listing 4-15, you can run it as many times as you want with the same parameters.

java –jar sample-application-0.0.1-SNAPSHOT.jar jobs/sampleJob.xml
sampleJob name=Michael -next

In fact, go ahead and give it a try. When you've run the sampleJob three or four times, look in the batch_job_params table and see how Spring Batch is executing your job with two parameters: one String named name with the value Michael, and one long named run.id. run.id's value changes each time, increasing by one with each execution.

You saw earlier that you may want to have a parameter be a timestamp with each run of the job. This is common in jobs that run once a day. To do so, you need to create your own implementation of JobParametersIncrementer. The configuration and execution are the same as before. However, instead of using RunIdIncrementer, you use DailyJobTimestamper, the code for which is in Listing 4-16.

package com.apress.springbatch.chapter4;

import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.JobParametersBuilder;
import org.springframework.batch.core.JobParametersIncrementer;

import java.util.Date;

import org.apache.commons.lang.time.DateUtils;

public class DailyJobTimestamper implements JobParametersIncrementer {

/**
* Increment the current.date parameter.
*/
public JobParameters getNext( JobParameters parameters ) {
Date today = new Date();

if ( parameters != null && !parameters.isEmpty() ) {
Date oldDate = parameters.getDate( "current.date", new Date() );
today = DateUtils.addDays(oldDate, 1);
}

return new JobParametersBuilder().addDate( "current.date", today )
                                 .toJobParameters();
}
}

It's pretty obvious that job parameters are an important part of the framework. They allow you to specify values at runtime for your job. They also are used to uniquely identify a run of your job. You use them more throughout the book for things like configuring the dates for which to run the job and reprocessing error files. For now, let's look at another powerful feature at the job level: job listeners.

As your jobs process, Spring Batch persists your state as part of committing each chunk. Part of that persistence is the saving of the job and current step's ExecutionContexts. Chapter 2 went over the layout of the tables. Let's go ahead and execute the sampleJob job with the updates from Listing 4-21 to see what the values look like persisted in the database. Table 4-2 shows what the batch_job_execution_context table has in it after a single run with the name parameter set as Michael.

{"map":{"entry":{"string":["user.name",
"Michael"]}}}

Table 4-2 consists of three columns. The first is a reference to the JobExecution that this ExecutionContext is related to. The second is a JSON representation of the Job's ExecutionContext. This field is updated as processing occurs. Finally, the SERIALIZED_CONTEXT field contains a serialized Java object. The SERIALIZED_CONTEXT is only populated while a job is running or when it has failed.

This section of the chapter has gone through different pieces of what a job is in Spring Batch. In order for a job to be valid, however, it requires at least one step, which brings you to the next major piece of the Spring Batch framework: steps.

When you think about steps in Spring Batch, there are two different types: a step for chunk-based processing and a tasklet step. Although you used a tasklet step previously in the "Hello, World!" job, you see more detail about it later. For now, you start by looking at how to configure chunk-based steps.

As you saw earlier, chunks are defined by their commit intervals. If the commit interval is set to 50 items, then your job reads in 50 items, processes 50 items, and then writes out 50 items at once. Because of this, the transaction manager plays a key part in the configuration of a chunk-based step. Listing 4-23 shows how to configure a basic step for chunk-oriented processing.

<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/batch"
    xmlns:beans="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://www.springframework.org/schema/beans
        http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
        http://www.springframework.org/schema/batch
        http://www.springframework.org/schema/batch/spring-batch-2.1.xsd">

    <beans:import resource="../launch-context.xml"/>

    <beans:bean id="inputFile"
        class="org.springframework.core.io.FileSystemResource" scope="step">
        <beans:constructor-arg value="#{jobParameters[inputFile]}"/>
    </beans:bean>

    <beans:bean id="outputFile"
        class="org.springframework.core.io.FileSystemResource" scope="step">
        <beans:constructor-arg value="#{jobParameters[outputFile]}"/>
    </beans:bean>

    <beans:bean id="inputReader"
        class="org.springframework.batch.item.file.FlatFileItemReader">
        <beans:property name="resource" ref="inputFile"/>
        <beans:property name="lineMapper">
            <beans:bean
  class="org.springframework.batch.item.file.mapping.PassThroughLineMapper"/>
        </beans:property>
    </beans:bean>

    <beans:bean id="outputWriter"
        class="org.springframework.batch.item.file.FlatFileItemWriter">
        <beans:property name="resource" ref="outputFile"/>
        <beans:property name="lineAggregator">
            <beans:bean
class="org.springframework.batch.item.file.transform.PassThroughLineAggregator"/>
        </beans:property>
    </beans:bean>

    <job id="stepJob">
        <step id="step1">
            <tasklet>
                <chunk reader="inputReader" writer="outputWriter"
                       commit-interval="50"/>
            </tasklet>
        </step>
    </job>
</beans:beans>

Listing 4-23 may look intimidating, but let's focus on the job and step configuration at the end. The rest of the file is the configuration of a basic ItemReader and ItemWriter, which are covered in Chapters 7 and 9, respectively. When you look through the job in Listing 4-23, you see that the step begins with the step tag. All that is required is the id or name, like any other Spring Bean. Within the step tag is a tasklet tag. The org.springframework.batch.core.step.tasklet.Tasklet interface is really a strategy interface for the type of step you're going to execute. In this case, you're configuring org.springframework.batch.core.step.item.ChunkOrientedTasklet<I>. You don't have to worry about configuring the class specifically here; just be aware that other types of tasklets can be used. The last piece of the example step is the chunk tag. Here you're defining what a chunk is for your step. You're saying to use the inputReader bean (an implementation of the ItemReader interface) as the reader and the outputWriter bean (an implementation of the ItemWriter interface) as the writer, and that a chunk consists of 50 items.

It's important to note the commit-interval attribute. It's set at 50 in the example. This means no records will be written until 50 records are read and processed. If an error occurs after processing 49 items, Spring Batch will roll back the current chunk (transaction) and mark the job as failed. If you were to set the commit-interval value to 1, your job would read in a single item, process that item, and then write that item. Essentially, you would be going back to item based processing. The issue with this is that there is more than just that single item being persisted at the commit-interval. The state of the job is being updated in the job repository as well. You experiment with the commit-interval later in this book but you needed to know now that it's important to set commit-interval as high as reasonably possible.

org.springframework.batch.core.step.tasklet.CallableTaskletAdapter is an adapter that allows you to configure an implementation of the java.util.concurrent.Callable<RepeatStatus> interface. If you're unfamiliar with this newer interface, the Callable<V> interface is similar to the java.lang.Runnable interface in that it's intended to be run in a new thread. However, unlike the Runnable interface, which doesn't return a value and can't throw checked exceptions, the Callable interface can return a value (a RepeatStatus, in this case) and can throw checked exceptions.

The adapter is actually extremely simple in its implementation. It calls the call() method on your Callable object and returns the value that the call() method returns. That's it. Obviously you would use this if you wanted to execute the logic of your step in another thread than the thread in which the step is being executed. If you look at Listing 4-24, you can see that to use this adapter, you configure CallableTaskletAdapter as a normal Spring bean and then reference it in the tasklet tag. In the configuration of the CallableTaskletAdapter bean shown in Listing 4-24, CallableTaskletAdapter contains a single dependency: the callable object itself.

<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/batch"
xmlns:beans="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://www.springframework.org/schema/beans
        http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
        http://www.springframework.org/schema/batch
        http://www.springframework.org/schema/batch/spring-batch-2.1.xsd">

    <beans:import resource="../launch-context.xml"/>

    <beans:bean id="callableObject"
        class="com.apress.springbatch.chapter4.CallableLogger"/>

    <beans:bean id="callableTaskletAdapter"
  class="org.springframework.batch.core.step.tasklet.CallableTaskletAdapter">
        <beans:property name="callable" ref="callableObject"/>
    </beans:bean>

    <job id="callableJob">
        <step id="step1">
            <tasklet ref="callableTaskletAdapter"/>
        </step>
    </job>
</beans:beans>

One thing to note with CallableTaskletAdapter is that although the tasklet is executed in a different thread than the step itself, this doesn't parallelize your step execution. The execution of this step won't be considered complete until the Callable object returns a valid RepeatStatus object. Until this step is considered complete, no other steps in the flow in which this step is configured will execute. You see how to parallelize processing in a number of ways, including executing steps in parallel, later in this book.

The next Tasklet implementation is org.springframework.batch.core.step.tasklet.MethodInvokingTaskletAdapter. This class is similar to a number of utility classes available in the Spring framework. It allows you to execute a preexisting method on another class as the step of your job. Say for example you already have a service that does a piece of logic that you want to run once in your batch job. Instead of writing an implementation of the Tasklet interface that really just wraps that method call, you can use MethodInvokingTaskletAdapter to call the method. Listing 4-25 shows an example of the configuration for MethodInvokingTaskletAdapter.

<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/batch"
    xmlns:beans="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://www.springframework.org/schema/beans
        http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
        http://www.springframework.org/schema/batch
        http://www.springframework.org/schema/batch/spring-batch-2.1.xsd">

    <beans:import resource="../launch-context.xml"/>

    <beans:bean id="service"
        class="com.apress.springbatch.chapter4.ChapterFourService"/>

    <beans:bean id="methodInvokingTaskletAdapter"
class="org.springframework.batch.core.step.tasklet.MethodInvokingTaskletAdapter">
        <beans:property name="targetObject" ref="service"/>
        <beans:property name="targetMethod" value="serviceMethod"/>
    </beans:bean>

    <job id="methodInvokingJob">
        <step id="step1">
            <tasklet ref="methodInvokingTaskletAdapter"/>
        </step>
    </job>
</beans:beans>

The example shown in Listing 4-25 specifies an object and a method. With this configuration, the adapter calls the method with no parameters and returns an ExitStatus.COMPLETED result unless the method specified also returns the type org.springframework.batch.core.ExitStatus. If it does return an ExitStatus, the value returned by the method is returned from the tasklet. If you want to configure a static set of parameters, you can use the late-binding method of passing job parameters that you read about earlier in this chapter, as shown in Listing 4-26.

beans:bean id="methodInvokingTaskletAdapter"
class="org.springframework.batch.core.step.tasklet.MethodInvokingTaskletAdapter"
        scope="step">
        <beans:property name="targetObject" ref="service"/>
        <beans:property name="targetMethod" value="serviceMethod"/>
        <beans:property name="arguments" value="#{jobParameters[message]}"/>
    </beans:bean>

    <job id="methodInvokingJob">
        <step id="step1">
            <tasklet ref="methodInvokingTaskletAdapter"/>
        </step>
    </job>

</beans:beans>

The last type of Tasklet implementation that Spring Batch provides is org.springframework.batch.core.step.tasklet.SystemCommandTasklet. This tasklet is used to—you guessed it—execute a system command! The system command specified is executed asynchronously. Because of this, the timeout value (in milliseconds) as shown in Listing 4-27 is important. The interruptOnCancel attribute in the listing is optional but indicates to Spring Batch whether to kill the thread the system process is associated with if the job exits abnormally.

<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/batch"
             xmlns:beans="http://www.springframework.org/schema/beans"
             xmlns:xsi=http://www.w3.org/2001/XMLSchema-instance
             xsi:schemaLocation="http://www.springframework.org/schema/beans
           http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
           http://www.springframework.org/schema/batch
           http://www.springframework.org/schema/batch/spring-batch-2.1.xsd">

    <beans:import resource="../launch-context.xml" />

    <beans:bean id="tempFileDeletionCommand"
    class="org.springframework.batch.core.step.tasklet.SystemCommandTasklet">
        <beans:property name="command" value="rm – rf /temp.txt " />
        <beans:property name="timeout" value="5000" />
        <beans:property name="interruptOnCancel" value="true" />
    </beans:bean>

    <job id="systemCommandJob">
        <step id="step1">
            <tasklet ref="tempFileDeletionCommand" />
        </step>
    </job>
</beans:beans>

SystemCommandTasklet allows you to configure a number of parameters that can have an effect on how a system command executes. Listing 4-28 shows a more robust example.

<beans:bean id="touchCodeMapper"
class="org.springframework.batch.core.step.tasklet.SimpleSystemProcessExitCodeMapper"/>

    <beans:bean id="taskExecutor"
        class="org.springframework.core.task.SimpleAsyncTaskExecutor"/>

    <beans:bean id="robustFileDeletionCommand"
    class="org.springframework.batch.core.step.tasklet.SystemCommandTasklet">

<beans:property name="command" value="touch temp.txt" />
        <beans:property name="timeout" value="5000" />
        <beans:property name="interruptOnCancel" value="true" />
        <beans:property name="workingDirectory"
            value="/Users/mminella/spring-batch" />
        <beans:property name="systemProcessExitCodeMapper"
            ref="touchCodeMapper"/>
        <beans:property name="terminationCheckInterval" value="5000" />
        <beans:property name="taskExecutor" ref="taskExecutor" />
        <beans:property name="environmentParams"
            value="JAVA_HOME=/java,BATCH_HOME=/Users/batch" />
    </beans:bean>

    <job id="systemCommandJob">
        <step id="step1">
            <tasklet ref="robustFileDeletionCommand" />
        </step>
    </job>
</beans:beans>

Listing 4-28 includes five more optional parameters in the configuration:

You've seen over the previous section that many different tasklet types are available in Spring Batch. Before moving off the topic, however, there is one other tasklet type to discuss: the tasklet step.

The tasklet step is different than the others you've seen. But it should be the most familiar to you, because it's what you used in the "Hello, World!" job. The way it's different is that in this case, you're writing your own code to be executed as the tasklet. Using MethodInvokingTaskletAdapter is one way to define a tasklet step. In that case, you allow Spring to forward the processing to your code. This lets you develop regular POJOs and use them as steps.

The other way to create a tasklet step is to implement the Tasklet interface as you did when you created the HelloWorld tasklet in Chapter 2. There, you implement the execute method required in the interface and return a RepeatStatus object to tell Spring Batch what to do after you completed processing. Listing 4-29 has the HelloWorld tasklet code as you constructed it in Chapter 2.

package com.apress.springbatch.chapter2;

import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;

public class HelloWorld implements Tasklet {

    private static final String HELLO_WORLD = "Hello, world!";

    public RepeatStatus execute( StepContribution arg0,
                                 ChunkContext arg1 ) throws Exception {
        System.out.println( HELLO_WORLD );
        return RepeatStatus.FINISHED;
    }
}

When processing is complete in your Tasklet implementation, you return an org.springframework.batch.repeat.RepeatStatus object. There are two options with this: RepeatStatus.CONTINUABLE and RepeatStatus.FINISHED. These two values can be confusing at first glance. If you return RepeatStatus.CONTINUABLE, you aren't saying that the job can continue. You're telling Spring Batch to run the tasklet again. Say, for example, that you wanted to execute a particular tasklet in a loop until a given condition was met, yet you still wanted to use Spring Batch to keep track of how many times the tasklet was executed, transactions, and so on. Your tasklet could return RepeatStatus.CONTINUABLE until the condition was met. If you return RepeatStatus.FINISHED, that means the processing for this tasklet is complete (regardless of success) and to continue with the next piece of processing.

You configure a tasklet step as you configure any of the other tasklet types. Listing 4-30 shows HelloWorldJob configured using the HelloWorld tasklet.

<beans:bean id="helloWorld" class="com.apress.springbatch.chapter2.HelloWorld/>

    <job id="helloWorldJob">
        <step id="helloWorldStep">
            <tasklet ref="helloWorld"/>
        </step>
    </job>
...

You may be quick to point out that this listing isn't the same as it was in Chapter 2, and you would be correct. The reason is that you haven't seen one other feature used in Chapter 2: step inheritance.

Like jobs, steps can be inherited from each other. Unlike jobs, steps don't have to be abstract to be inherited. Spring Batch allows you to configure fully defined steps in your configuration and then have other steps inherit them. Let's start discussing step inheritance by looking at the example used in Chapter 2, the HelloWorldJob in Listing 4-31.

<beans:bean id="helloWorld"
    class="com.apress.springbatch.chapter2.HelloWorld"/>

<step id="helloWorldStep">
    <tasklet ref="helloWorld"/>
</step>

<job id="helloWorldJob">
    <step id="step1" parent="helloWorldStep"/>
</job>

In Listing 4-31, you configure the tasklet implementation (the helloWorld bean), and then you configure the step that references the tasklet (helloWorldStep). Spring Batch doesn't require that the step element be nested in a job tag. Once you're defined your step, helloWorldStep, you can then inherit it when you declare the steps in sequence in your actual job, helloWorldJob. Why would you do this?

In this simple example, there is little benefit to this approach. However, as steps become more complex, experience shows that it's best to configure your steps outside the scope of your job and then inherit them with the steps in the job. This allows the actual job declaration to be much more readable and maintainable.

Obviously, readability isn't the only reason to use inheritance, and that isn't all that is going on even in this example. Let's dive deeper. In this example, what you're really doing in step1 is inheriting the step helloWorldStep and all its attributes. However, step1 chooses not to override any of them.

Step inheritance provides a more complete inheritance model than that of job inheritance. In step inheritance you can fully define a step, inherit the step, and then add or override any of the values you wish. You can also declare a step abstract and place only common attributes there.

Listing 4-32 shows an example of how steps can add and override attributes configured in their parent. You start with the parent step, vehicleStep, which declares a reader, writer, and commit-interval. You then create two steps that inherit from vehicleStep: carStep and truckStep. Each uses the same reader and writer that has been configured in vehicleStep. In each case, they add an item processor that does different things. carStep has chosen to use the inherited commit-interval of 50 items, whereas truckStep has overridden the commit-interval and set it to 5 items.

<step id="vehicleStep">
<tasklet>
<chunk reader="vehicleReader" writer="vehicleWriter" commit-interval="50"/>
</tasklet>
</step>

<step id="carStep" parent="vehicleStep">
<tasklet>
<chunk processor="carProcessor"/>
</tasklet>
</step>

<step id="truckStep" parent="vehicleStep">
<tasklet>
<chunk processor="truckProcessor" commit-interval="5"/>
</tasklet>
</step>

<job id="exampleJob">
<step id="step1" parent="carStep" next="step2"/>
<step id="step2" parent="truckStep"/>
</job>

By declaring a step abstract, as in Java, you're allowed to leave things out that would otherwise be required. In an abstract step, as in Listing 4-33, you're allowed to leave off the reader, writer, processor, and tasklet attributes. This would normally cause an initialization error when Spring tried to build the step; but because it's declared abstract, Spring knows that those will be populated by the steps that inherit it.

<beans:bean id="inputFile"
        class="org.springframework.core.io.FileSystemResource" scope="step">
        <beans:constructor-arg value="#{jobParameters[inputFile]}"/>
    </beans:bean>

    <beans:bean id="outputFile"
        class="org.springframework.core.io.FileSystemResource" scope="step">
        <beans:constructor-arg value="#{jobParameters[outputFile]}"/>
    </beans:bean>

    <beans:bean id="inputReader"
        class="org.springframework.batch.item.file.FlatFileItemReader">
        <beans:property name="resource" ref="inputFile"/>
        <beans:property name="lineMapper">

<beans:bean
 class="org.springframework.batch.item.file.mapping.PassThroughLineMapper"/>
        </beans:property>
    </beans:bean>

    <beans:bean id="outputWriter"
        class="org.springframework.batch.item.file.FlatFileItemWriter">
        <beans:property name="resource" ref="outputFile"/>
        <beans:property name="lineAggregator">
            <beans:bean
class="org.springframework.batch.item.file.transform.PassThroughLineAggregator"/>
        </beans:property>
    </beans:bean>

    <step id="commitIntervalStep" abstract="true">
        <tasklet>
            <chunk commit-interval="15"/>
        </tasklet>
    </step>

    <step id="copyStep" parent="commitIntervalStep">
        <tasklet>
            <chunk reader="inputReader" writer="outputWriter" />
        </tasklet>
    </step>

    <job id="stepInheritanceJob">
        <step id="step1" parent="copyStep" />
    </job>
</beans:beans>

In Listing 4-33, commitIntervalStep is an abstract step that is used to configure the commit interval for any step that extends this step. You configure the required elements of a step in the step that extends the abstract step, copyStep. Here you specify a reader and writer. copyStep has the same commit-interval of 15 that commitIntervalStep has, without the need to repeat the configuration.

Step inheritance allows you to configure common attributes that can be reused from step to step as well as structure your XML configuration in a maintainable way. The last example of this section used a couple of attributes that were chunk specific. To better understand them, let's go over how you can use the different features that Spring Batch provides in its chunk-based processing.

Because chunk-based processing is the foundation of Spring Batch 2, it's important to understand how to configure its various options to take full advantage of this important feature. This section covers the two options for configuring the size of a chunk: a static commit count and a CompletionPolicy implementation. All other chunk configuration options relate to error handling and are discussed in that section.

To start looking at chunk configuration, Listing 4-34 has a basic example of nothing more than a reader, writer, and commit-interval configured. The reader is an implementation of the ItemReader interface, and the writer an implementation of ItemWriter. Each of these interfaces has its own dedicated chapter later in the book, so this section doesn't go into detail about them. All you need to know is that they supply input and output, respectively, for the step. The commit-interval defines how many items make up a chunk (50 items, in this case).

<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/batch"
             xmlns:beans="http://www.springframework.org/schema/beans"
             xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xsi:schemaLocation="http://www.springframework.org/schema/beans
             http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
             http://www.springframework.org/schema/batch
           http://www.springframework.org/schema/batch/spring-batch-2.1.xsd">

    <beans:import resource="../launch-context.xml" />

    <beans:bean id="inputFile"
        class="org.springframework.core.io.FileSystemResource" scope="step">
        <beans:constructor-arg value="#{jobParameters[inputFile]}"/>
    </beans:bean>

    <beans:bean id="outputFile"
        class="org.springframework.core.io.FileSystemResource" scope="step">
        <beans:constructor-arg value="#{jobParameters[outputFile]}"/>
    </beans:bean>

    <beans:bean id="inputReader"
        class="org.springframework.batch.item.file.FlatFileItemReader">
        <beans:property name="resource" ref="inputFile"/>
        <beans:property name="lineMapper">
            <beans:bean
class="org.springframework.batch.item.file.mapping.PassThroughLineMapper"/>
        </beans:property>
    </beans:bean>

    <beans:bean id="outputWriter"
        class="org.springframework.batch.item.file.FlatFileItemWriter">
        <beans:property name="resource" ref="outputFile"/>
        <beans:property name="lineAggregator">
            <beans:bean
class="org.springframework.batch.item.file.transform.PassThroughLineAggregator"/>
        </beans:property>
    </beans:bean>

    <step id="copyStep">
        <tasklet>
            <chunk reader="inputReader" writer="outputWriter"
                commit-interval="50"/>
        </tasklet>
    </step>

    <job id="chunkConfigurationJob">

<step id="step1" parent="copyStep" />
    </job>
</beans:beans>

Although typically you define the size of a chunk based on a hard number configured with the commit-interval attribute as configured in Listing 4-34, that isn't always a robust enough option. Say that you have a job that needs to process chunks that aren't all the same size (processing all transactions for an account in a single transaction, for example). Spring Batch provides the ability to programmatically define when a chunk is complete via an implementation of the org.springframework.batch.repeat.CompletionPolicy interface.

The CompletionPolicy interface allows the implementation of decision logic to decide if a given chunk is complete. Spring Batch comes with a number of implementations of this interface. By default it uses org.springframework.batch.repeat.policy.SimpleCompletionPolicy, which counts the number of items processed and flags a chunk complete when the configured threshold is reached. Another out-of-the-box implementation is org.springframework.batch.repeat.policy.TimeoutTerminationPolicy. This allows you to configure a timeout on a chunk so that it may exit gracefully after a given amount of time. What does "exit gracefully" mean in this context? It means that the chunk is considered complete and all transaction processing continues normally.

As you can undoubtedly deduce, there are few times when a timeout by itself is enough to determine when a chunk of processing will be complete. TimeoutTerminationPolicy is more likely to be used as part of org.springframework.batch.repeat.policy.CompositeCompletionPolicy. This policy lets you configure multiple polices that determine whether a chunk has completed. When you use CompositeCompletionPolicy, if any of the policies consider a chunk complete, then the chunk is flagged as complete. Listing 4-35 shows an example of using a timeout of 3 milliseconds along with the normal commit count of 200 items to determine if a chunk is complete.

<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/batch"

    xmlns:beans="http://www.springframework.org/schema/beans"
    xmlns:util="http://www.springframework.org/schema/util"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://www.springframework.org/schema/beans
        http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
        http://www.springframework.org/schema/util
        http://www.springframework.org/schema/util/spring-util.xsd
        http://www.springframework.org/schema/batch
        http://www.springframework.org/schema/batch/spring-batch-2.1.xsd">

    <beans:import resource="../launch-context.xml" />

    <beans:bean id="inputFile"
        class="org.springframework.core.io.FileSystemResource" scope="step">
        <beans:constructor-arg value="#{jobParameters[inputFile]}" />
    </beans:bean>

    <beans:bean id="outputFile"
        class="org.springframework.core.io.FileSystemResource" scope="step">
        <beans:constructor-arg value="#{jobParameters[outputFile]}" />
    </beans:bean>

<beans:bean id="inputReader"
        class="org.springframework.batch.item.file.FlatFileItemReader">
        <beans:property name="resource" ref="inputFile" />
        <beans:property name="lineMapper">
            <beans:bean
class="org.springframework.batch.item.file.mapping.PassThroughLineMapper" />
        </beans:property>
    </beans:bean>

    <beans:bean id="outputWriter"
        class="org.springframework.batch.item.file.FlatFileItemWriter">
        <beans:property name="resource" ref="outputFile" />
        <beans:property name="lineAggregator">
            <beans:bean
class="org.springframework.batch.item.file.transform.PassThroughLineAggregator" />
        </beans:property>
    </beans:bean>

    <beans:bean id="chunkTimeout"
    class="org.springframework.batch.repeat.policy.TimeoutTerminationPolicy">
        <beans:constructor-arg value="3" />
    </beans:bean>

    <beans:bean id="commitCount"
      class="org.springframework.batch.repeat.policy.SimpleCompletionPolicy">
        <beans:property name="chunkSize" value="200" />
    </beans:bean>

    <beans:bean id="chunkCompletionPolicy"
   class="org.springframework.batch.repeat.policy.CompositeCompletionPolicy">
        <beans:property name="policies">
            <util:list>
                <beans:ref bean="chunkTimeout" />
                <beans:ref bean="commitCount" />
            </util:list>
        </beans:property>
    </beans:bean>

    <step id="copyStep">
        <tasklet>
            <chunk reader="inputReader" writer="outputWriter"
                    chunk-completion-policy="chunkCompletionPolicy"/>
        </tasklet>
    </step>

    <job id="chunkConfigurationJob">
        <step id="step1" parent="copyStep" />
    </job>
</beans:beans>

Using the implementations of the CompletionPolicy interface isn't your only option to determine how large a chunk is. You can also implement it yourself. Before you look at an implementation, let's go over the interface.

The CompletionPolicy interface requires four methods: two versions of isComplete, start, and update. If you look at this through the lifecycle of the class, first the start method is called first. This method initializes the policy so that it knows the chunk is starting. It's important to note that an implementation of the CompletionPolicy interface is intended to be stateful and should be able to determine if a chunk has been completed by its own internal state. The start method resets this internal state to whatever is required by the implementation at the beginning of the chunk. Using SimpleCompletionPolicy as an example, the start method resets an internal counter to 0 at the beginning of a chunk. The update method is called once for each item that has been processed to update the internal state. Going back to the SimpleCompletionPolicy example, update increments the internal counter by one after each item. Finally, there are two isComplete methods. The first isComplete method signature accepts a RepeatContext as its parameter. This implementation is intended to use its internal state to determine if the chunk has completed. The second signature takes the RepeatContext and also the RepeatStatus as parameters. This implementation is expected to determine based on the status whether a chunk has completed. Listing 4-36 shows an example of a CompletionPolicy implementation that considers a chunk complete once a random number of items fewer than 20 have been processed; Listing 4-37 showing the configuration.

package com.apress.springbatch.chapter4;

import java.util.Random;

import org.springframework.batch.repeat.CompletionPolicy;
import org.springframework.batch.repeat.RepeatContext;
import org.springframework.batch.repeat.RepeatStatus;

public class RandomChunkSizePolicy implements CompletionPolicy {

    private int chunkSize;
    private int totalProcessed;

    public boolean isComplete(RepeatContext context) {
        return totalProcessed >= chunkSize;
    }

    public boolean isComplete(RepeatContext context, RepeatStatus status) {
        if (RepeatStatus.FINISHED == status) {
            return true;
        } else {
            return isComplete(context);
        }
    }

    public RepeatContext start(RepeatContext context) {
        Random random = new Random();

        chunkSize = random.nextInt(20);

totalProcessed = 0;

        System.out.println("The chunk size has been set to " + chunkSize);

        return context;
    }

    public void update(RepeatContext context) {
        totalProcessed++;
    }
}

<beans:bean id="randomChunkSizer"
    class="com.apress.springbatch.chapter4.RandomChunkSizePolicy" />

<step id="copyStep">
    <tasklet>
        <chunk reader="inputReader" writer="outputWriter"
            chunk-completion-policy="randomChunkSizer" />
    </tasklet>
</step>

<job id="chunkConfigurationJob">
    <step id="step1" parent="copyStep" />
</job>

You explore the rest of chunk configuration when you get to error handling. That section covers retry and skip logic, which the majority of the remaining options center around. The next step elements that this chapter looks at also carry over from a job: listeners.

When you looked at job listeners, earlier this chapter, you saw the two events they can fire on: the start and end of a job. Step listeners cover the same types of events (start and end), but for individual steps instead of an entire job. This section covers the org.springframework.batch.core.StepExecutionListener and org.springframework.batch.core.ChunkListener interfaces, both of which allow the processing of logic at the beginning and end of a step and chunk respectively. Notice that the Step's listener is named the StepExecutionListener and not just StepListener. There actually is a StepListener interface, however it's just a marker interface that all step related listeners extend.

Both the StepExecutionListener and ChunkListener provide methods that are similar to the ones in the JobExecutionListener interface. StepExecutionListener has a beforeStep and an afterStep, and ChunkListener has a beforeChunk and an afterChunk, as you would expect. All of these methods are void except afterStep. afterStep returns an ExitStatus because the listener is allowed to modify the ExitStatus that was returned by the step itself prior to it being returned to the job. This feature can be useful when a job requires more than just knowing whether an operation was successful to determine if the processing was successful. An example would be doing some basic integrity checks after importing a file (whether the correct number of records were written to the database, and so on). The ability to configure listeners via annotations also continues to be consistent, with Spring Batch providing @BeforeStep, @AfterStep, @BeforeChunk, and @AfterChunk annotations to simplify the implementation. Listing 4-38 shows a StepListener that uses annotations to identify the methods.

package com.apress.springbatch.chapter4;

import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.annotation.AfterStep;
import org.springframework.batch.core.annotation.BeforeStep;

public class LoggingStepStartStopListener {

    @BeforeStep
    public void beforeStep(StepExecution execution) {
        System.out.println(execution.getStepName() + " has begun!");
    }

    @AfterStep
    public ExitStatus afterStep(StepExecution execution) {
        System.out.println(execution.getStepName() + " has ended!");

       return execution.getExitStatus();
    }
}

The configuration for all the step listeners is combined into a single list in the step configuration. Similar to the job listeners, inheritance works the same way, allowing you to either override the list or merge them together. Listing 4-39 configures the LoggingStepStartStopListener that you coded earlier.

...
<beans:bean id="loggingStepListener"
    class="com.apress.springbatch.chapter4.LoggingStepStartStopListener"/>

<job id="stepListenerJob">
    <step id="step1">
        <tasklet>
            <chunk reader="inputReader" writer="outputWriter"
                commit-interval="50"/>
            <listeners>
                <listener ref="loggingStepListener"/>
            </listeners>
        </tasklet>
    </step>
</job>
...

As you can see, listeners are available at just about every level of the Spring Batch framework to allow you to hang processing off your batch jobs. They're commonly used not only to perform some form of preprocessing before a component or evaluate the result of a component but also in error handling, as you see in a bit.

The next section covers the flow of steps. Although all your steps up to this point have been processed sequentially, that isn't a requirement in Spring Batch. You learn how to perform simple logic to determine what step to execute next and how to externalize flows for reuse.

Within a job in Spring Batch, steps are executed in the order you specify using the next attribute of the step tag. The only requirement is that the first step be configured as the first step in the job. If you want to execute steps in a different order, it's quite easy: all you need to do is use the next tag. As Listing 4-40 shows, you can use the next tag to direct a job to go from step1 to step2a if things go ok or to step2b if step1 returns an ExitStatus of FAILED.

<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/batch"
    xmlns:beans="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://www.springframework.org/schema/beans
        http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
        http://www.springframework.org/schema/batch
        http://www.springframework.org/schema/batch/spring-batch-2.1.xsd">

    <beans:import resource="../launch-context.xml"/>

    <beans:bean id="passTasklet"
        class="com.apress.springbatch.chapter4.LogicTasklet">
        <beans:property name="success" value="true"/>
    </beans:bean>

    <beans:bean id="successTasklet"
        class="com.apress.springbatch.chapter4.MessageTasklet">
        <beans:property name="message" value="The step succeeded!"/>
    </beans:bean>

    <beans:bean id="failTasklet"
        class="com.apress.springbatch.chapter4.MessageTasklet">
        <beans:property name="message" value="The step failed!"/>
    </beans:bean>

<job id="conditionalStepLogicJob">
        <step id="step1">
            <tasklet ref="passTasklet"/>
            <next on="*" to="step2a"/>
            <next on="FAILED" to="step2b"/>
        </step>
        <step id="step2a">
            <tasklet ref="successTasklet"/>
        </step>
        <step id="step2b">
            <tasklet ref="failTasklet"/>
        </step>
    </job>
</beans:beans>

The next tag uses the on attribute to evaluate the ExitStatus of the step and determine what to do. It's important to note that you've seen both org.springframework.batch.core.ExitStatus and org.springframework.batch.core.BatchStatus over the course of this chapter. BatchStatus is an attribute of the JobExecution or StepExecution that identifies the current state of the job or step. ExitStatus is the value returned to Spring Batch at the end of a job or step. The on attribute evaluates the ExitStatus for its decisions. So, the example in Listing 4-40 is the XML equivalent of saying, "If the exit code of step1 doesn't equal FAILED, go to step2a, else go to step2b."

Because the values of the ExitStatus are really just Strings, the ability to use wildcards can make things interesting. Spring Batch allows for two wildcards in on criteria:

Although evaluating the ExitStatus gets you started in determining what to do next, it may not take you all the way. For example, what if you didn't want to execute a step if you skipped any records in the current step? You wouldn't know that from the ExitStatus alone.

Spring Batch has provided a programmatic way to determine what to do next. You do this by creating an implementation of the org.springframework.batch.core.job.flow.JobExecutionDecider interface. This interface has a single method, decide, that takes both the JobExecution and the StepExecution and returns a FlowExecutionStatus (a wrapper for a BatchStatus/ExitStatus pair). With both the JobExecution and StepExecution available for evaluation, all information should be available to you to make the appropriate decision about what your job should do next. Listing 4-41 shows an implementation of the JobExecutionDecider that randomly decides what the next step should be.

package com.apress.springbatch.chapter4;

import java.util.Random;

import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.job.flow.FlowExecutionStatus;
import org.springframework.batch.core.job.flow.JobExecutionDecider;

public class RandomDecider implements JobExecutionDecider {

    private Random random = new Random();

    public FlowExecutionStatus decide(JobExecution jobExecution,
            StepExecution stepExecution) {

        if (random.nextBoolean()) {
            return new
                FlowExecutionStatus(FlowExecutionStatus.COMPLETED.getName());
        } else {
            return new
                FlowExecutionStatus(FlowExecutionStatus.FAILED.getName());
        }
    }
}

To use RandomDecider, you configure an extra attribute on your step called decider. This attribute refers to the Spring bean that implements JobExecutionDecider. Listing 4-42 shows RandomDecider configured. You can see that the configuration maps the values you return in the decider to steps available to execute.

...
<beans:bean id="decider"
    class="com.apress.springbatch.chapter4.RandomDecider"/>

<beans:bean id="successTasklet"
    class="com.apress.springbatch.chapter4.MessageTasklet">
    <beans:property name="message" value="The step succeeded!"/>
</beans:bean>

<beans:bean id="failTasklet"
    class="com.apress.springbatch.chapter4.MessageTasklet">
    <beans:property name="message" value="The step failed!"/>
</beans:bean>

<job id="conditionalLogicJob">
    <step id="step1" next="decision">
        <tasklet>
            <chunk reader="inputReader" writer="outputWriter"
                commit-interval="20"/>

</tasklet>
    </step>
    <decision decider="decider" id="decision">
        <next on="*" to="step2a"/>
        <next on="FAILED" to="step2b"/>
    </decision>
    <step id="step2a">
        <tasklet ref="successTasklet"/>
    </step>
    <step id="step2b">
        <tasklet ref="failTasklet"/>
    </step>
</job>
...

Because you now know how to direct your processing from step to step either sequentially or via logic, you won't always want to just go to another step. You may want to end or pause the job. The next section covers how to handle those scenarios.

You learned earlier that a JobInstance can't be executed more than once to a successful completion and that a JobInstance is identified by the job name and the parameters passed into it. Because of this, you need to be aware of the state in which you end your job if you do it programmatically. In reality, there are three states in which you can programmatically end a job in Spring Batch:

It's important to note that these states are identified by Spring Batch evaluating the ExitStatus of the step to determine what BatchStatus to persist in the JobRepository. ExitStatus can be returned from a step, chunk, or job. BatchStatus is maintained in StepExecution or JobExecution and persisted in the JobRepository. Let's begin looking at how to end the job in each state with the completed state.

To configure a job to end in the completed state based on the exit status of a step, you use the end tag. In this state, you can't execute the same job again with the same parameters. Listing 4-43 shows that the end tag has a single attribute that declares the ExitStatus value that triggers the job to end.

<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/batch"
    xmlns:beans="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

xsi:schemaLocation="http://www.springframework.org/schema/beans
        http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
        http://www.springframework.org/schema/batch
        http://www.springframework.org/schema/batch/spring-batch-2.1.xsd">

    <beans:import resource="../launch-context.xml"/>

    <beans:bean id="passTasklet"
        class="com.apress.springbatch.chapter4.LogicTasklet">
        <beans:property name="success" value="false"/>
    </beans:bean>

    <beans:bean id="successTasklet"
        class="com.apress.springbatch.chapter4.MessageTasklet">
        <beans:property name="message" value="The step succeeded!"/>
    </beans:bean>

    <beans:bean id="failTasklet"
        class="com.apress.springbatch.chapter4.MessageTasklet">
        <beans:property name="message" value="The step failed!"/>
    </beans:bean>

    <job id="conditionalStepLogicJob">
        <step id="step1">
            <tasklet ref="passTasklet"/>
            <end on="*"/>
            <next on="FAILED" to="step2b"/>
        </step>
        <step id="step2b">
            <tasklet ref="failTasklet"/>
        </step>
    </job>
</beans:beans>

Once you run conditionalStepLogicJob, as you would expect, the batch_step_execution table contains the ExitStatus returned by the step, and batch_job_execution contains COMPLETED regardless of the path taken.

For the failed state, which allows you to rerun the job with the same parameters, the configuration looks similar. Instead of using the end tag, you use the fail tag. Listing 4-44 shows that the fail tag has an additional attribute: exit-code. It lets you add extra detail when causing a job to fail.

<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/batch"
    xmlns:beans="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://www.springframework.org/schema/beans
        http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
        http://www.springframework.org/schema/batch
        http://www.springframework.org/schema/batch/spring-batch-2.1.xsd">

<beans:import resource="../launch-context.xml"/>

    <beans:bean id="passTasklet"
        class="com.apress.springbatch.chapter4.LogicTasklet">
        <beans:property name="success" value="true"/>
    </beans:bean>

    <beans:bean id="successTasklet"
        class="com.apress.springbatch.chapter4.MessageTasklet">
        <beans:property name="message" value="The step succeeded!"/>
    </beans:bean>

    <beans:bean id="failTasklet"
        class="com.apress.springbatch.chapter4.MessageTasklet">
        <beans:property name="message" value="The step failed!"/>
    </beans:bean>

    <job id="conditionalStepLogicJob">
        <step id="step1">
            <tasklet ref="passTasklet"/>
            <next on="*" to="step2a"/>
            <fail on="FAILED" exit-code="STEP-1-FAILED"/>
        </step>
        <step id="step2a">
            <tasklet ref="successTasklet"/>
        </step>
    </job>
</beans:beans>

When you rerun conditionalStepLogicJob with the configuration in Listing 4-44, the results are a bit different. This time, if step1 ends with the ExitStatus FAILURE, the job is identified in the jobRepository as failed, which allows it to be reexecuted with the same parameters.

The last state you can leave a job in when you end it programmatically is the stopped state. In this case, you can restart the job; and when you do, it restarts at the step you configure. Listing 4-45 shows an example.

<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/batch"
    xmlns:beans="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://www.springframework.org/schema/beans
        http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
        http://www.springframework.org/schema/batch
        http://www.springframework.org/schema/batch/spring-batch-2.1.xsd">

    <beans:import resource="../launch-context.xml"/>

    <beans:bean id="passTasklet"
        class="com.apress.springbatch.chapter4.LogicTasklet">
        <beans:property name="success" value="true"/>

</beans:bean>

    <beans:bean id="successTasklet"
        class="com.apress.springbatch.chapter4.MessageTasklet">
        <beans:property name="message" value="The step succeeded!"/>
    </beans:bean>

    <beans:bean id="failTasklet"
        class="com.apress.springbatch.chapter4.MessageTasklet">
        <beans:property name="message" value="The step failed!"/>
    </beans:bean>

    <job id="conditionalStepLogicJob">
        <step id="step1">
            <tasklet ref="passTasklet"/>
            <next on="*" to="step2a"/>
            <stop on="FAILED" restart="step2a"/>
        </step>
        <step id="step2a">
            <tasklet ref="successTasklet"/>
        </step>
    </job>
</beans:beans>

Executing conditionalStepLogicJob with this final configuration, as in Listing 4-45, allows you to rerun the job with the same parameters. However, this time, if the FAILURE path is chosen, when the job is restarted execution begins at step2a.

The flow from one step to the next isn't just another layer of configuration you're adding to potentially complex job configurations; it's also configurable in a reusable component. The next section discusses how to encapsulate flows of steps into reusable components.

You've already identified that a step doesn't need to be configured within a job tag in your XML. This lets you extract the definition of your steps from a given job into reusable components. The same goes for the order of steps. In Spring Batch, there are three options for how to externalize the order of steps. The first is to create a flow, which is an independent sequence of steps. The second is to use the flow step; although the configuration is very similar, the state persistence in the JobRepository is slightly different. The last way is to actually call another job from within your job. This section covers how all three of these options work.

A flow looks a lot like a job. It's configured the same way, but with a flow tag instead of a job tag. Listing 4-46 shows how to define a flow using the flow tag, giving it an id and then referencing it in your job using the flow tag.

<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/batch"
    xmlns:beans="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://www.springframework.org/schema/beans

http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
        http://www.springframework.org/schema/batch
        http://www.springframework.org/schema/batch/spring-batch-2.1.xsd">

    <beans:import resource="../launch-context.xml"/>

    <beans:bean id="loadStockFile"
        class="com.apress.springbatch.chapter4.MessageTasklet">
        <beans:property name="message"
            value="The stock file has been loaded"/>
    </beans:bean>

    <beans:bean id="loadCustomerFile"
        class="com.apress.springbatch.chapter4.MessageTasklet">
        <beans:property name="message"
            value="The customer file has been loaded" />
    </beans:bean>

    <beans:bean id="updateStart"
        class="com.apress.springbatch.chapter4.MessageTasklet">
        <beans:property name="message"
            value="The stock file has been loaded" />
    </beans:bean>

    <beans:bean id="runBatchTasklet"
        class="com.apress.springbatch.chapter4.MessageTasklet">
        <beans:property name="message" value="The batch has been run" />
    </beans:bean>

    <flow id="preProcessingFlow">
        <step id="loadFileStep" next="loadCustomerStep">
            <tasklet ref="loadStockFile"/>
        </step>
        <step id="loadCustomerStep" next="updateStartStep">
            <tasklet ref="loadCustomerFile"/>
        </step>
        <step id="updateStartStep">
            <tasklet ref="updateStart"/>
        </step>
    </flow>

    <job id="flowJob">
        <flow parent="preProcessingFlow" id="step1" next="runBatch"/>
        <step id="runBatch">
            <tasklet ref="runBatchTasklet"/>
        </step>
    </job>
</beans:beans>

When you execute a flow as part of a job and look at the jobRepository, you see the steps from the flow recorded as part of the job as if they were configured there in the first place. In the end, there is no difference between using a flow and configuring the steps within the job itself from a JobRepository perspective.

The next option for externalizing steps is to use the flow step. With this technique, the configuration of a flow is the same. But instead of using the flow tag to include the flow in your job, you use a step tag and its flow attribute. Listing 4-47 demonstrates how to use a flow step to configure the same example Listing 4-46 used.

...
<flow id="preProcessingFlow">
    <step id="loadFileStep" next="loadCustomerStep">
        <tasklet ref="loadStockFile"/>
    </step>
    <step id="loadCustomerStep" next="updateStartStep">
        <tasklet ref="loadCustomerFile"/>
    </step>
    <step id="updateStartStep">
        <tasklet ref="updateStart"/>
    </step>
</flow>

<job id="flowJob">
    <step id="initializeBatch" next="runBatch">
        <flow parent="preProcessingFlow"/>
    </step>
    <step id="runBatch">
        <tasklet ref="runBatchTasklet"/>
    </step>
</job>
...

What is the difference between using the flow tag and the flow step? It comes down to what happens in the JobRepository. Using the flow tag ends up with the same results as if you configured the steps in your job. Using a flow step adds an additional entry. When you use a flow step, Spring Batch records the step that includes the flow as a separate step. Why is this a good thing? The main benefit is for monitoring and reporting purposes. Using a flow step allows you to see the impact of the flow as a whole instead of having to aggregate the individual steps.

The last way to externalize the order in which steps occur is to not externalize them at all. In this case, instead of creating a flow, you call a job from within another job. Similar to the flow step, which creates a StepExecutionContext for the execution of the flow and each step within it, the job step creates a JobExecutionContext for the step that calls the external job. Listing 4-48 shows the configuration of a job step.

<job id="preProcessingJob">
    <step id="step1" parent="loadStockFile" next="step2"/>
    <step id="step2" parent="loadCustomerFile" next="step3"/>
    <step id="step3" parent="updateStartOfBatchCycle"/>
</job>

<beans:bean id="jobParametersExtractor"
class="org.springframework.batch.core.step.job.DefaultJobParametersExtractor">

<beans:property name="keys" value="job.stockFile,job.customerFile"/>
</beans:bean>

<job id="subJobJob">
    <step id="step0" next="step4 ">
        <job ref="preProcessingJob"
            job-parameters-extractor="jobParametersExtractor "/>
    </step>
    <step id="step4" parent="runBatch"/>
</job>

You might be wondering about the jobParametersExtractor bean in Listing 4-48. When you launch a job, it's identified by the job name and the job parameters. In this case, you aren't passing the parameters to your sub job, preProcessingJob, by hand. Instead, you define a class to extract the parameters from either the JobParameters of the parent job or the ExecutionContext (the DefaultJobParameterExtractor checks both places) and pass those parameters to the child job. Your extractor pulls the values from the job.stockFile and job.customerFile job parameters and passes those as parameters to preProcessingJob.

When preProcessingJob executes, it's identified in the JobRepository just like any other job. It has its own job instance, execution context, and related database records.

A word of caution about using the job-step approach: this may seem like a good way to handle job dependencies. Creating individual jobs and being able to then string them together with a master job is a powerful feature. However, this can severely limit the control of the process as it executes. It isn't uncommon in the real world to need to pause a batch cycle or skip jobs based on external factors (another department can't get you a file in time to have the process finished in the required window, and so on). However, the ability to manage jobs exists at a single job level. Managing entire trees of jobs that could be created using this functionality is problematic and should be avoided. Linking jobs together in this manner and executing them as one master job severely limits the capability to handle these types of situations and should also be avoided.

The last piece of the flow puzzle is the ability Spring Batch provides to execute multiple flows in parallel, which is covered next.

Although you learn about parallelization later in this book, this section covers the Spring Batch functionality to execute step flows in parallel. One of the strengths of using Java for batch processing and the tools that Spring Batch provides is the ability to bring multithreaded processing to the batch world in a standardized way. One of the easiest ways to execute steps in parallel is to split them in your job.

The split flow is a step that allows you to list flows that you want to execute in parallel. Each flow is started at the same time, and the step isn't considered complete until all the flows within it have completed. If any one of the flows fails, the split flow is considered to have failed. To see how the split step works, look at Listing 4-49.

<job id="flowJob">
    <split id="preprocessingStep" next="batchStep">
        <flow>
            <step id="step1" parent="loadStockFile" next="step2"/>
            <step id="step2" parent="loadCustomerFile"/>

</flow>
        <flow>
            <step id="step3" parent="loadTransactionFile"/>
        </flow>
    </split>
    <step id="batchStep" parent="runBatch"/>
</job>

In Listing 4-49, you identify two separate flows: one that loads two files and one that loads one file. Each of these flows is executed in parallel. After all three steps (step1, step2, and step3) have completed, Spring Batch executes batchStep.

That's it. It's amazing how simple it is to do basic parallelization using Spring Batch, as this example shows. And given the potential performance boosts,^[14] you can begin to see why Spring Batch can be a very effective tool in high-performance batch processing.

Later chapters cover a variety of error-handling scenarios including error handling at the entire job level down to errors at the transaction level. But because steps are all about processing chunks of items, the next topic is some of the error-handling strategies available when processing a single item.

When you're processing large amounts of data, it isn't uncommon to have errors due to things that don't need human intervention. If a database that is shared across systems has a deadlock, or a web service call fails due to a network hiccup, stopping the processing of millions of items is a pretty drastic way to handle the situation. A better approach is to allow your job to try to process a given item again.

There are three ways to implement retry logic in Spring Batch: configuring retry attempts, using RetryTemplate, and using Spring's AOP features. The first approach lets Spring Batch define a number of allowed retry attempts and exceptions that trigger the new attempt. Listing 4-50 show the basic retry configuration for an item when a RemoteAccessException is thrown in the course of executing the remoteStep.

<job id="flowJob">
    <step id="retryStep">
        <tasklet>
            <chunk reader="itemReader" writer="itemWriter"
                processor="itemProcessor" commit-interval="20"
                retry-limit="3">
                <retryable-exception-classes>
                    <include
                class="org.springframework.remoting.RemoteAccessException"/>
                </retryable-exception-classes>
            </chunk>
        </tasklet>
    </step>
</job>

In the flowJob's retryStep, when a RemoteAccessException is thrown by any of the components in the step (itemReader, itemWriter, or itemProcessor) the item is retried up to three times before the step fails.

Another way to add retry logic to your batch job is to do it yourself via org.springframework.batch.retry.RetryTemplate. Like most other templates provided in Spring, this one simplifies the development of retry logic by providing a simple API to encapsulate the retryable logic in a method that is then managed by Spring. In the case of RetryTemplate, you need to develop two pieces: the org.springframework.batch.retry.RetryPolicy interface and the org.springframework.batch.retry.RetryCallback interface. RetryPolicy allows you to define under what conditions an item's processing is to be retried. Spring provides a number of implementations, including ones for retrying based on an exception being thrown (which you used by default in Listing 4-50), timeout, and others. The other piece of coding retry logic is the use of the RetryCallback interface. This interface provides a single method, doWithRetry(RetryContext context), that encapsulates the logic to be retried. When you use RetryTemplate, if an item is to be retried, the doWithRetry method is called as long as RetryPolicy specifies it to be. Let's look at an example.

Listing 4-51 shows the code to retry a call to a database with a timeout policy of 30 seconds. This means it will continue to try executing the database call until it works or until 30 seconds has passed.

package com.apress.springbatch.chapter4;

import java.util.List;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.retry.support.RetryTemplate;
import org.springframework.batch.retry.RetryCallback;
import org.springframework.batch.retry.RetryContext;

public class RetryItemWriter implements ItemWriter<Customer> {

    private CustomerDAO customerDao;
    private RetryTemplate retryTemplate;

    public void write(List<? extends Customer> customers) throws Exception {

for (Customer customer : customers) {
            final Customer curCustomer = customer;

            retryTemplate.execute(new RetryCallback<Customer>() {
                public Customer doWithRetry(RetryContext retryContext) {
                    return customerDao.save(curCustomer);
                }
            });
        }
    }
   ...
}

The code in Listing 4-51 depends significantly on elements that need to be injected (getters and setters were left out of the example). To get a better view of what is happening here, Listing 4-52 shows the configuration for this example.

<beans:bean id="timeoutPolicy"
    class="org.springframework.batch.retry.policy.TimeoutRetryPolicy">
    <beans:property name="timeout" value="30000"/>
</beans:bean>

<beans:bean id="timeoutRetryTemplate"
    class="org.springframework.batch.retry.support.RetryTemplate">
    <beans:property name="retryPolicy" ref="timeoutPolicy"/>
</beans:bean>

<beans:bean id="retryItemWriter"
    class="com.apress.springbatch.chapter4.RetryItemWriter">
    <beans:property name="customerDao" ref="customerDao"/>
    <beans:property name="retryTemplate" ref="timeoutRetryTemplate"/>
</beans:bean>

<job id="flowJob">
    <step id="retryStep">
        <tasklet>
            <chunk reader="itemReader" writer="retryItemWriter"
                    processor="itemProcessor" commit-interval="20"/>
        </tasklet>
    </step>
</job>

Most of the configuration in Listing 4-52 should be straightforward. You configure the org.springframework.batch.retry.policy.TimeoutRetryPolicy bean with a timeout value set to 30 seconds; you inject that into RetryTemplate as the retryPolicy and inject the template into the ItemWriter you wrote for Listing 4-51. One thing that is interesting about the configuration of this retry, however, is that there is no retry configuration in the job. Because you write your own retry logic, you don't use the retry-limit and so on in the chunk configuration.

The final way to configure item-retry logic is to use Spring's AOP facilities and Spring Batch's org.springframework.batch.retry.interceptor.RetryOperationsInterceptor to declaratively apply retry logic to elements of your batch job. Listing 4-53 shows how to declare the aspect's configuration to apply retry logic to any save methods.

<aop:config>
    <aop:pointcut id="saveRetry"
    expression="execution(* com.apress.springbatch.chapter4.*.save(..))"/>
    <aop:advisor pointcut-ref="saveRetry" advice-ref="retryAdvice"
        order="-1"/>
</aop:config>

<beans:bean id="retryAdvice"

    class="org.springframework.batch.retry.interceptor.RetryOperationsInterceptor"/>

This configuration leaves out the definition of how many retries to attempt, and so on. To add those, all you need to do is configure the appropriate RetryPolicy into RetryOperationsInterceptor (which has an optional dependency for a RetryPolicy implementation).

RetryPolicy is similar to CompletionPolicy earlier in that it allows you to programmatically decide something—in this case, when to retry. The org.springbatch.retry.RetryPolicy used in either the AOP interceptor or in the regular retry approaches is important, but one thing to note is that you may not want to just retry over and over if it isn't working. For example, when Google's Gmail can't connect back to the server, it first tries to reconnect immediately, then it waits 15 seconds, then 30 seconds, and so on. This approach can prevent multiple retries from stepping on each other's toes. Fortunately, Spring Batch provides a BackoffPolicy interface to implement this type of decay. You can implement an algorithm yourself by implementing the BackoffPolicy interface or use the ExponentialBackOffPolicy provided by the framework. Listing 4-54 shows the configuration required for BackOffPolicy.

<beans:bean id="timeoutPolicy"
    class="org.springframework.batch.retry.policy.TimeoutRetryPolicy">
    <beans:property name="timeout" value="30000"/>
</beans:bean>

<beans:bean id="backoutPolicy"

                     class="org.springframework.batch.retry.backoff.ExponentialBackOffPolicy"/>

<beans:bean id="timeoutRetryTemplate"
    class="org.springframework.batch.retry.support.RetryTemplate">
    <beans:property name="retryPolicy" ref="timeoutPolicy"/>
    <beans:property name="backOffPolicy" ref="backOffPolicy"/>
</beans:bean>

<beans:bean id="retryItemWriter"
    class="com.apress.springbatch.chapter4.RetryItemWriter">
    <beans:property name="customerDao" ref="customerDao"/>
    <beans:property name="retryTemplate" ref="timeoutRetryTemplate"/>
</beans:bean>

<job id="flowJob">

<step id="retryStep">
        <tasklet>
            <chunk reader="itemReader" writer="retryItemWriter"
                    processor="itemProcessor"
                     commit-interval="20"/>
        </tasklet>
    </step>
</job>

The last aspect of retry logic is that, like most available events in Spring Batch, retry has the ability to register listeners to when an item is being retried. There are two differences, however, between all the other listeners and org.springframework.batch.retry.RetryListener. First, RetryListener has no annotation equivalent of the interface, so if you want to register a listener on retry logic, you have to implement the RetryListener interface. The other difference is that instead of two methods in the interface for the start and end, there are three in this interface. In RetryListener, the open method is called when the retry block is about to be called, onError is called once for each retry, and close is called when the full retry block is complete.

That covers it for retry logic. The other way to handle item-specific error handling is to skip the item altogether.

One of the greatest things in Spring Batch is the ability to skip an item that is causing problems. This feature can easily prevent a phone call in the middle of the night to deal with a production problem if the item can be addressed the next day. Configuring the ability to skip an item is similar to configuring retry logic. All you need to do is use the skip-limit attribute on the chunk tag and specify the exceptions that should cause an item to be skipped. Listing 4-55 demonstrates how to configure a step to allow a maximum of 10 items to be skipped via skip-limit. It then states that any item that causes any subclass of java.lang.Exception except for java.lang.NullPointerException is allowed to be skipped. Any item that throws a NullPointerException causes the step to end in error.

<job id="flowJob">
    <step id="retryStep">
        <tasklet>
            <chunk reader="itemReader" writer="itemWriter"
                processor="itemProcessor" commit-interval="20"
                skip-limit="10">
                <skippable-exception-classes>
                    <include class="java.lang.Exception"/>
                    <exclude class="java.lang.NullPointerException"/>
                </skippable-exception-classes>
            </chunk>
        </tasklet>
    </step>
</job>

When using item-based error handling, whether it's retrying to process an item or skipping it, there can be transactional implications. You learn what those are and how to address them when the book gets into reading and writing items in Chapters 7 & 9.