Spring Batch - Reference Documentation

The FieldSetMapper interface defines a single method, mapLine, which takes a FieldSet object and maps its contents to an object. This object may be a custom DTO or domain object, or it could be as simple as an array, depending on your needs. The FieldSetMapper is used in conjunction with the LineTokenizer to translate a line of data from a resource into an object of the desired type:

  public interface FieldSetMapper {
  
    public Object mapLine(FieldSet fs);

  }

The pattern used is the same as RowMapper used by JdbcTemplate.

Because there can be many formats of flat file data, which all need to be converted to a FieldSet so that a FieldSetMapper can create a useful domain object from them, an abstraction for turning a line of input into a FieldSet is necessary. In Spring Batch, this is called a LineTokenizer:

  public interface LineTokenizer {
  
    FieldSet tokenize(String line);

  }

The contract of a LineTokenizer is such that, given a line of input (in theory the String could encompass more than one line) a FieldSet representing the line will be returned. This will then be passed to a FieldSetMapper. Spring Batch contains the following LineTokenizers:

Now that the basic interfaces for reading in flat files have been defined, a simple example explaining how they work together is helpful. In it's most simple form, the flow when reading a line from a file is the following:

In code, the above flow looks like the following:

  String line = readLine();

  if (line != null) {
    FieldSet tokenizedLine = tokenizer.tokenize(line);
    return fieldSetMapper.mapLine(tokenizedLine);
  }
 
  return null;

The following example will be used to illustrate this using an actual domain scenario. This particular batch job reads in football players from the following file:

  ID,lastName,firstName,position,birthYear,debutYear
  "AbduKa00,Abdul-Jabbar,Karim,rb,1974,1996",
  "AbduRa00,Abdullah,Rabih,rb,1975,1999",
  "AberWa00,Abercrombie,Walter,rb,1959,1982",
  "AbraDa00,Abramowicz,Danny,wr,1945,1967",
  "AdamBo00,Adams,Bob,te,1946,1969",
  "AdamCh00,Adams,Charlie,wr,1979,2003"        

The contents of this file will be mapped to the following Player domain object:

  public class Player implements Serializable {
        
  private String ID; 
  private String lastName; 
  private String firstName; 
  private String position; 
  private int birthYear; 
  private int debutYear;
        
    public String toString() {
                
      return "PLAYER:ID=" + ID + ",Last Name=" + lastName + 
        ",First Name=" + firstName + ",Position=" + position + 
        ",Birth Year=" + birthYear + ",DebutYear=" + 
        debutYear;
    }
   
    // setters and getters...
  }
          

In order to map a FieldSet into a Player object, a FieldSetMapper that returns players needs to be defined:

  protected static class PlayerFieldSetMapper implements FieldSetMapper {
    public Object mapLine(FieldSet fieldSet) {
      Player player = new Player();

      player.setID(fieldSet.readString(0));
      player.setLastName(fieldSet.readString(1));
      player.setFirstName(fieldSet.readString(2)); 
      player.setPosition(fieldSet.readString(3));
      player.setBirthYear(fieldSet.readInt(4));
      player.setDebutYear(fieldSet.readInt(5));

      return player;
    }
  }    

The file can then be read by correctly constructing a FlatFileItemReader and calling read:

  
  FlatFileItemReader itemReader = new FlatFileItemReader();
  itemReader.setResource(new FileSystemResource("resources/players.csv"));
  //DelimitedLineTokenizer defaults to comma as it's delimiter
  itemReader.setLineTokenizer(new DelimitedLineTokenizer());
  itemReader.setFieldSetMapper(new PlayerFieldSetMapper());
  itemReader.open(new ExecutionContext());
  Player player = (Player)itemReader.read();

Each call to read will return a new Player object from each line in the file. When the end of the file is reached, null will be returned.

There is one additional functionality a LineTokenizer that is similar in function to a JDBC ResultSet. The names of the fields can be injected into the LineTokenizer to increase the readability of the mapping function. First, the column names of all fields in the flat file are injected into the LineTokenizer:

  tokenizer.setNames(new String[] {"ID", "lastName","firstName","position","birthYear","debutYear"}); 
          

a FieldSetMapper can this use this information as follows:

    public class PlayerMapper implements FieldSetMapper {
        public Object mapLine(FieldSet fs) {
                        
           if(fs == null){
              return null;
           }
                        
           Player player = new Player();
           player.setID(fs.readString("ID"));
           player.setLastName(fs.readString("lastName"));
           player.setFirstName(fs.readString("firstName"));
           player.setPosition(fs.readString("position"));
           player.setDebutYear(fs.readInt("debutYear"));
           player.setBirthYear(fs.readInt("birthYear"));
                        
           return player;
        }

   }        

For many, having to write a specific FieldSetMapper is equally as cumbersome as writing a specific RowMapper for a JdbcTemplate. Spring Batch makes this easier by providing a FieldSetMapper that automatically maps fields by matching a field name with a setter on the object using the JavaBean spec. Again using the football example, the FieldSetMapper configuration looks like the following:

  <bean id="fieldSetMapper"
        class="org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper">
    <property name="prototypeBeanName" value="player" />
  </bean>

  <bean id="player"
        class="org.springframework.batch.sample.domain.Player"
        scope="prototype" />

For each entry in the FieldSet, the mapper will look for a corresponding setter on a new instance of the Player object (for this reason, prototype scope is required) in the same way the Spring container will look for setters matching a property name. Each available field in the FieldSet will be mapped, and the resultant Player object will be returned, with no code required.

So far only delimited files have been discussed in much detail, however, they respresent only half of the file reading picture. Many organizations that use flat files use fixed length formats. An example fixed length file is below:

  UK21341EAH4121131.11customer1
  UK21341EAH4221232.11customer2
  UK21341EAH4321333.11customer3
  UK21341EAH4421434.11customer4
  UK21341EAH4521535.11customer5

While this looks like one large field, it actually represent 4 distinct fields:

When configuring the FixedLengthLineTokenizer, each of these lengths must be provided in the form of ranges:

  
  <bean id="fixedLengthLineTokenizer"
        class="org.springframework.batch.item.file.transform.FixedLengthTokenizer">
    <property name="names" value="ISIN, Quantity, Price, Customer" />
    <property name="columns" value="1-12, 13-15, 16-20, 21-29" />
  </bean>

This LineTokenizer will return the same FieldSet as if a dlimiter had been used, allowing the same approachs above to be used such as the BeanWrapperFieldSetMapper, in a way that is ignorant of how the actual line was parsed.

It should be noted that supporting the above ranges requires a specialized property editor be configured anywhere in the ApplicationContext:

  
  <bean id="customEditorConfigurer" class="org.springframework.beans.factory.config.CustomEditorConfigurer">
    <property name="customEditors">
      <map>
        <entry key="org.springframework.batch.item.file.transform.Range[]">
          <bean class="org.springframework.batch.item.file.transform.RangeArrayPropertyEditor" />
        </entry>
      </map>
    </property>
  </bean>

All of the file reading examples up to this point have all made a key assumption for simplicity's sake: one record equals one line. However, this may not always be the case. Its very common that a file might have records spanning multiple lines with multiple formats. The following excerpt from a file illustrates this:

  HEA;0013100345;2007-02-15
  NCU;Smith;Peter;;T;20014539;F
  BAD;;Oak Street 31/A;;Small Town;00235;IL;US
  SAD;Smith, Elizabeth;Elm Street 17;;Some City;30011;FL;United States
  BIN;VISA;VISA-12345678903
  LIT;1044391041;37.49;0;0;4.99;2.99;1;45.47
  LIT;2134776319;221.99;5;0;7.99;2.99;1;221.87
  SIN;UPS;EXP;DELIVER ONLY ON WEEKDAYS
  FOT;2;2;267.34

Everything between the line starting with 'HEA' and the line starting with 'FOT' is considered one record. The PrefixMatchingCompositeLineTokenizer makes this easier by matching the prefix in a line with a particular tokenizer:

  <bean id="orderFileDescriptor"
        class="org.springframework.batch.item.file.transform.PrefixMatchingCompositeLineTokenizer">
    <property name="tokenizers">
     <map>
      <entry key="HEA" value-ref="headerRecordDescriptor" />
      <entry key="FOT" value-ref="footerRecordDescriptor" />
      <entry key="BCU" value-ref="businessCustomerLineDescriptor" />
      <entry key="NCU" value-ref="customerLineDescriptor" />
      <entry key="BAD" value-ref="billingAddressLineDescriptor" />
      <entry key="SAD" value-ref="shippingAddressLineDescriptor" />
      <entry key="BIN" value-ref="billingLineDescriptor" />
      <entry key="SIN" value-ref="shippingLineDescriptor" />
      <entry key="LIT" value-ref="itemLineDescriptor" />
      <entry key="" value-ref="defaultLineDescriptor" />
     </map>
    </property>
  </bean>

This ensures that the line will be parsed correctly, which is especially important for fixed length input. Any users of the FlatFileItemReader in this scenario must continue calling read until the footer for the record is returned, allowing them to return a complete order as one 'item'.

Just as the LineTokenizer interface is necessary to take a string and split it into tokens, file writing must have a way to aggregate multiple fields into a single string for writing to a file. In Spring Batch this is the LineAggregator:

  public interface LineAggregator {

    public String aggregate(FieldSet fieldSet);

  }

The LineAggregator is the opposite of a LineTokenizer. LineTokenizer takes a String and returns a FieldSet, whereas LineAggregator takes a FieldSet and returns a String. As with reading there are two types: DelimitedLineAggregator and FixedLengthLineAggregator.

Because the LineAggregator interface uses a FieldSet as it's mechanism for converting to a string, there needs to be an interface that describes how to convert from an object into a FieldSet:

  public interface FieldSetCreator {

    FieldSet mapItem(Object data);

  }

As with LineTokenizer and LineAggregator, FieldSetCreator is the polar opposite of FieldSetMapper. FieldSetMapper takes a FieldSet and returns a mapped object, whereas a FieldSetCreator takes an Object and returns a FieldSet.

Now that both the LineAggregator and FieldSetCreator interfaces have been defined, the basic flow of writing can be explained:

The following excerpt from the FlatFileItemWriter expresses this in code:

  public void write(Object data) throws Exception {
    FieldSet fieldSet = fieldSetCreator.mapItem(data);
    getOutputState().write(lineAggregator.aggregate(fieldSet) + LINE_SEPARATOR);
  }

A simple configuration with the smallest ammount of setters would look like the following:

  <bean id="itemWriter"
        class="org.springframework.batch.item.file.FlatFileItemWriter">
    <property name="resource"
              value="file:target/test-outputs/20070122.testStream.multilineStep.txt" />
    <property name="fieldSetCreator">
      <bean class="org.springframework.batch.item.file.mapping.PassThroughFieldSetMapper"/>
    </property>
  </bean>

FlatFileItemReader has a very simple relationship with file resources. When the reader is initialized, it opens the file if it exists, and throws an exception if it does not. File writing isn't quite so simple. At first glance it seems like a similar straight forward contract should exist for FlatFileItemWriter, if the file already exists, throw an exception, if it does not, create it and start writing. However, potentially restarting a Job can cause issues. In normal restart scenarios, the contract is reversed, if the file exists start writing to it from the last known good position, if it does not, throw an exception. However, what happens if the file name for this job is always the same? In this case, you would want to delete the file if it exists, unless it's a restart. Because of this possibility, the FlatFileItemWriter contains the property, shouldDeleteIfExists. Setting this property to true will cause an existing file with the same name to be deleted when the writer is opened.

JdbcCursorItemReader is the JDBC implementation of the cursor based technique. It works directly with a ResultSet and requires a SQL statement to run against a connection obtained from a DataSource. The following database schema will be used as an example:

CREATE TABLE CUSTOMER (
 ID BIGINT IDENTITY PRIMARY KEY,  
 NAME VARCHAR(45),
 CREDIT FLOAT
);

Many people prefer to use a domain object for each row, so we'll use an implementation of the RowMapper interface to map a CustomerCredit object:

public class CustomerCreditRowMapper implements RowMapper {

 public static final String ID_COLUMN = "id";
 public static final String NAME_COLUMN = "name";
 public static final String CREDIT_COLUMN = "credit";

 public Object mapRow(ResultSet rs, int rowNum) throws SQLException {
        CustomerCredit customerCredit = new CustomerCredit();

        customerCredit.setId(rs.getInt(ID_COLUMN));
        customerCredit.setName(rs.getString(NAME_COLUMN));
        customerCredit.setCredit(rs.getBigDecimal(CREDIT_COLUMN));

        return customerCredit;
 }

}

Because JdbcTemplate is so familiar to users of Spring, and the JdbcCursorItemReader shares key interfaces with it, it's useful to see an example of how to read in this data with JdbcTemplate, in order to contrast it with the ItemReader. For the purposes of this example, let's assume there are 1,000 rows in the CUSTOMER database. The first example will be using JdbcTemplate:

  
  //For simplicity sake, assume a dataSource has already been obtained
  JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);
  List customerCredits = jdbcTemplate.query("SELECT ID, NAME, CREDIT from CUSTOMER", new CustomerCreditRowMapper());

After running this code snippet the customerCredits list will contain 1,000 CustomerCredit objects. In the query method, a connection will be obtained from the DataSource, the provided SQL will be run against it, and the mapRow method will be called for each row in the ResultSet. Let's constrast this with the approach of the JdbcCursorItemReader:

  
  JdbcCursorItemReader itemReader = new JdbcCursorItemReader();
  itemReader.setDataSource(dataSource);
  itemReader.setSql("SELECT ID, NAME, CREDIT from CUSTOMER");
  itemReader.setMapper(new CustomerCreditRowMapper());
  int counter = 0;
  ExecutionContext executionContext = new ExecutionContext();
  itemReader.open(executionContext);
  Object customerCredit = new Object();
  while(customerCredit != null){
    customerCredit = itemReader.read();
    counter++;
  }
  itemReader.close(executionContext);

After running this code snippet the counter will equal 1,000. If the code above had put the returned customerCredit into a list, the result would have been exactly the same as with the JdbcTemplate example. However, the big advantage of the ItemReader is that it allows items to be 'streamed'. The read method can be called once, and the item written out via an ItemWriter, and then the next item obtained via read. This allows item reading and writing to be done in 'chunks' and committed periodically, which is the essence of high performance batch processing.

Just as normal Spring users make important decisions about whether or not to use ORM solutions, which affects whether or not they use a JdbcTemplate or a HibernateTemplate, Spring Batch users have the same options. HibernateCursorItemReader is the Hibernate implementation of the cursor technique. Hibernate's usage in batch has been fairly controversial. This has largely been because hibernate was originally developed to support online application styles. However, that doesn't mean it can't be used for batch processing. The easiest approach for solving this problem is to use a StatelessSession rather than a standard session. This removes all of the caching and dirty checking hibernate employs that can cause issues when using it in a batch scenario. For more information on the differences between stateless and normal hibernate sessions, refer to the documentation of your specific hibernate release. The HibernateCursorItemReader allows you to declare an HQL statement and pass in a SessionFactory, which will pass back one item per call to read in the same basic fashion as the JdbcCursorItemReader. Below is an example configuration using the same 'customer credit' example as the JDBC reader:

  
  HibernateCursorItemReader itemReader = new HibernateCursorItemReader();
  itemReader.setQueryString("from CustomerCredit");
  //For simplicity sake, assume sessionFactory already obtained.
  itemReader.setSessionFactory(sessionFactory);
  itemReader.setUseStatelessSession(true);
  int counter = 0;
  ExecutionContext executionContext = new ExecutionContext();
  itemReader.open(executionContext);
  Object customerCredit = new Object();
  while(customerCredit != null){
    customerCredit = itemReader.read();
    counter++;
  }
  itemReader.close(executionContext);

This configured ItemReader will return CustomerCredit objects in the exact same manner as described by the JdbcCursorItemReader, assuming hibernate mapping files have been created correctly for the Customer table. The 'useStatelessSession' property defaults to true, but has been added here to draw attention to the ability to switch it on or off. It is also worth noting that the fetchSize of the underlying cursor can be set via the setFetchSize property.

As the previous example illustrates, the DrivingQueryItemReader is fairly simple. It simply iterates over a list of keys. However, the real complication is how those keys are obtained. The KeyCollector interface abstracts this:

  public interface KeyCollector {

    List retrieveKeys(ExecutionContext executionContext);

    void updateContext(Object key, ExecutionContext executionContext);
  }

The primary method in this interface is the retrieveKeys method. It is expected that this method will return the keys to be processed regardless of whether or not it is a restart scenario. For example, if a job starts processing keys 1 through 1,000, and fails after processing key 500, upon restarting keys 500 through 1,000 should be returned. This functionality is made possible by the updateContext method, which saves the provided key (which should be the current key being processed) in the provided ExecutionContext. The retrieveKeys method can then use this value to retrieve a subset of the original keys:

  ExecutionContext executionContext = new ExecutionContext();
  List keys = keyStrategy.retrieveKeys(executionContext);
  //Assume keys contains 1 through 1,000
  keyStrategy.updateContext(new Long(500), executionContext);
  keys = keyStrategy.retrieveKeys(executionContext);
  //keys should now contains 500 through 1,000

This generalization illustrates the KeyCollector contract. If we assume that initially calling retrieveKeys returned 1,000 keys (1 through 1,000), calling updateContext with key 500 should mean that calling retrieveKeys again with the same ExecutionContext will return 500 keys (501 through 1,000).

The most common driving query scenario is that of input that has only one column that represents its key. This is implemented as the SingleColumnJdbcKeyCollector class, which has the following options:

The following code helps illustrate how to setup and use a SingleColumnJdbcKeyCollector:

  SingleColumnJdbcKeyCollector keyCollector = new SingleColumnJdbcKeyCollector(getJdbcTemplate(),
  "SELECT ID from T_FOOS order by ID");

  keyCollector.setRestartSql("SELECT ID from T_FOOS where ID > ? order by ID");

  ExecutionContext executionContext = new ExecutionContext();

  List keys = keyStrategy.retrieveKeys(new ExecutionContext());

  for (int i = 0; i < keys.size(); i++) {
    System.out.println(keys.get(i));
  }

If this code were run in the proper environment with the correct database tables setup, then it would output the following:

1
2
3
4
5

Now, let's modify the code slightly to show what would happen if the code were started again after a restart, having failed after processing key 3 successfully:

  SingleColumnJdbcKeyCollector keyCollector = new SingleColumnJdbcKeyCollector(getJdbcTemplate(),
  "SELECT ID from T_FOOS order by ID");

  keyCollector.setRestartSql("SELECT ID from T_FOOS where ID > ? order by ID");

  ExecutionContext executionContext = new ExecutionContext();  

  keyStrategy.updateContext(new Long(3), executionContext);
  
  List keys = keyStrategy.retrieveKeys(executionContext);

  for (int i = 0; i < keys.size(); i++) {
    System.out.println(keys.get(i));
  }

Running this code snippet would result in the following:

4
5

The key difference between the two examples is the following line:

  keyStrategy.updateContext(new Long(3), executionContext);

This tells the key collector to update the provided ExecutionContext with the key of three. This will normally be called by the DrivingQueryItemReader, but is called directly for simplicities sake. By calling retrieveKeys with the ExecutionContext that was updated to contain 3, the argument of 3 will be passed to the restartSql:

  keyCollector.setRestartSql("SELECT ID from T_FOOS where ID > ? order by ID");

This will cause only keys 4 and 5 to be returned, since they are the only ones with an ID greater than 3.

The SingleColumnJdbcKeyCollector is extremely useful for generating keys, but only if one column uniquely identifies your record. What if more than one column is required to be able to uniquely identify your record? This should be a minority scenario, but it is still possible. In this case, the MultipleColumnJdbcKeyCollector should be used. It allows for mapping multiple columns by sacrificing simplicity. The properties needed to use the multiple column collector are the same as the single column version except one difference: instead of a regular RowMaper, an ExecutionContextRowMapper must be provided. Just like the single column version, it requires a normal SQL statement and a restart SQL statement. However, because the restart SQL statement will require more than one argument, there needs to be more complex handling of how keys are mapped to an execution context. An ExecutionContextRowMapper provides this:

public interface ExecutionContextRowMapper extends RowMapper {

  public void mapKeys(Object key, ExecutionContext executionContext);

  public PreparedStatementSetter createSetter(ExecutionContext executionContext);
}

The ExecutionContextRowMapper interface extends the standard RowMapper interface to allow for multiple keys to be stored in an ExecutionContext, and a PreparedStatementSetter be created so that arguments to a the restart SQL statement can be set for the key returned.

By default a implementation of the ExecutionContextRowMapper that uses a Map will be used. It is recommended that this implementation not be overridden. However, if a specific type of key needs to be returned, then a new implementation can be provided.

Jdbc is not the only option available for key collectors, iBatis can be used as well. The usage of iBatis doesn't change the basic requirements of a KeyCollector: query, restart query, and DataSource. However, because iBatis is used, both queries are simply iBatis query ids, and the data source is a SqlMapClient.

Note that the ItemTransformerItemWriter and the CompositeItemWriter are examples of a delegation pattern, which is common in Spring Batch. The delegates themselves might implement callback interfaces like ItemStream or StepListener. If they do, and they are being used in conjunction with Spring Batch Core as part of a Step in a Job, then they almost certainly need to be registered manually with the Step. Registration is automatic when using the factory beans (*StepFactoryBean) , but only for the ItemReader and ItemWriter injected directly. The delegates are not known to the Step, so they need to be injected as listeners or streams (or both if appropriate).

Note that the ValidatingItemReader is another example of a delegation pattern, and the delegates themselves might implement callback interfaces like ItemStream or StepListener. If they do, and they are being used in conjunction with Spring Batch Core as part of a step in a job, then they almost certainly need to be registered manually with the Step. Registration is automatic when using the factory beans (*StepFactoryBean) , but only for the ItemReader and ItemWriter injected directly - the delegates are not known to the step, so they need to be injected as listeners or streams (or both if appropriate).

This most basic ItemReader will work, but what happens if the transaction needs to be rolled back? This will usually caused by an error in the ItemWriter, since the ItmReader generally won't do anything that invalidates the transaction, but without supporting it, there would be erroneous results. ItemReaders are notified about rollbacks via the mark and reset methods. In the example above they're empty, but we'll need to add code to them in order to support the rollback scenario:

  public class CustomItemReader implements ItemReader{

    List items;
    int currentIndex = 0;
    int lastMarkedIndex = 0;

    public CustomItemReader(List items) {
      this.items = items;
    }

    public Object read() throws Exception, UnexpectedInputException,
       NoWorkFoundException, ParseException {

      if (currentIndex < items.size()) {
        return items.get(currentIndex++);
      }
      return null;
    }

    public void mark() throws MarkFailedException {
      lastMarkedIndex = currentIndex;
    };

    public void reset() throws ResetFailedException {
      currentIndex = lastMarkedIndex;
    };
  }

The CustomItemReader has now been modified to keep track of where it is currently, and where it was when mark() was last called. This allows the new ItemReader to fulfill the basic contract that calling reset returns the ItemReader to the state it was in when mark was last called:

  //Assume same setup as last example, a list with "1", "2", and "3"
  itemReader.mark();
  assertEquals("1", itemReader.read());
  assertEquals("2", itemReader.read());
  itemReader.reset();
  assertEquals("1", itemReader.read());

In most real world scenarios, there will likely be some kind of underlying resource that will require tracking. In the case of a file, mark will hold the current location within the file, and reset will move it back. The JdbcCursorItemReader, for example, holds on to the current row number, and on reset moves the cursor back by calling the ResultSet absolute method, which moves the current cursor to the row number supplied. The CustomItemReader now completely adheres to the entire ItemReader contract. read will return the appropriates items, returning null when empty, and reset returns the ItemReader back to it's state as of the last call to mark, allowing for correct support of a rollback. (It's assumed a Step implementation will call mark and reset).

The final challenge now is to make the ItemReader restartable. Currently, if the power goes out, and processing begins again, the ItemReader must start at the beginning. This is actually valid in many scenarios, but it is sometimes preferable that a batch job starts off at where it left off. The key discriminant is often whether the reader is stateful or stateless. A stateless reader does not need to worry about restartablility, but a stateful one has to try and reconstitute its last known state on restart. For this reason, we recommend that you keep custom readers stateless as far as possible, so you don't have to worry about restartability.

If you do need to store state, then in Spring Batch, this is implemented with the ItemStream interface:

  public class CustomItemReader implements ItemReader, ItemStream{

    List items;
    int currentIndex = 0;
    int lastMarkedIndex = 0;
    private static String CURRENT_INDEX = "current.index";

    public CustomItemReader(List items) {
      this.items = items;
    }

    public Object read() throws Exception, UnexpectedInputException,
        NoWorkFoundException, ParseException {

      if (currentIndex < items.size()) {
        return items.get(currentIndex++);
      }
      return null; 
    }

    public void mark() throws MarkFailedException {
      lastMarkedIndex = currentIndex;
    };

    public void reset() throws ResetFailedException {
      currentIndex = lastMarkedIndex;
    }

    public void open(ExecutionContext executionContext) throws ItemStreamTException {
      if(executionContext.containsKey(CURRENT_INDEX)){
        currentIndex = new Long(executionContext.getLong(CURRENT_INDEX)).intValue();
      }
      else{
        currentIndex = 0;
        lastMarkedIndex = 0;
      }
    }

    public void update(ExecutionContext executionContext) throws ItemStreamException {
      executionContext.putLong(CURRENT_INDEX, new Long(currentIndex).longValue());
    };

    public void close(ExecutionContext executionContext) throws ItemStreamException {}
  }

On each call to ItemStream update method, the current index of the ItemReader will be stored in the provided ExecutionContext with a key of 'current.index'. When the ItemStream open method is called, the ExecutionContext is checked to see if it contains an entry with that key, and if so the current index is moved to that location. This is a fairly trivial example, but it still meets the general contract:

  ExecutionContext executionContext = new ExecutionContext();
  ((ItemStream)itemReader).open(executionContext);
  assertEquals("1", itemReader.read());
  ((ItemStream)itemReader).update(executionContext);

  List items = new ArrayList();
  items.add("1");
  items.add("2");
  items.add("3");
  itemReader = new CustomItemReader(items);

  ((ItemStream)itemReader).open(executionContext);
  assertEquals("2", itemReader.read());

Most ItemReaders have much more sophisticated restart logic. The DrivingQueryItemReader, for example, only loads up the remaining keys to be processed, rather than loading all of them and then moving to the correct index.

It is also worth noting that the key used within the ExecutionContext should not be trivial. That is because the same ExecutionContext is used for all ItemStreams within a Step. In most cases, simply prepending the key with the class name should be enough to guarantee uniqueness. However, in the rare cases where two of the same type of ItemStream are used in the same step (which can happen if two files are need for output) then a more unique name will be needed. For this reason, many of the Spring Batch ItemReader and ItemWriters have a setName() property that allows this key name to be overridden.

The example is extremely simple, but it's worth showing to illustrate an ItemWriter that doesn't respond to rollbacks and commits (i.e. clear and flush). If your potential writer is such that it doesn't need to care about rollback or commit, likely because it's writing to a database, then there is little value to the ItemWriter interface in that scenario other than using it to meet another class's requirement for an implementation of the ItemWriter interface. In that case, the ItemWriterAdapter would be a better solution. However, if it does need to be transactional, then flush and clear should be implemented to allow for a buffering solution:

  public class CustomItemWriter implements ItemWriter{

    List output = new ArrayList();
    List buffer = new ArrayList();

    public void write(Object item) throws Exception {
      buffer.add(item);
    }
  
    public void clear() throws ClearFailedException {
      buffer.clear();
    }

    public void flush() throws FlushFailedException {
      for(Iterator it = buffer.iterator(); it.hasNext();){
        output.add(it.next());
        it.remove();
      }
    }
  }

The ItemWriter buffers all output, only writing to the actual output (in this case by added to a list) when the ItemWriter flush() method is called. The contents of the buffer are thrown away when ItemWriter clear() is called.

To make the ItemWriter restartable we would follow the same process as for the ItemReader, adding and implementing the ItemStream interface to synchronize the execution context. In the example we might have to count the number of items processed and add that as a footer record. If we needed to do that, we could implement ItemStream in our ItemWriter so that the counter was reconstituted from the execution context if the stream was re-opened.

In many realistic cases, custom ItemWriters also delegate to another writer that itself is restartable (e.g. when writing to a file), or else it writes to a transactional resource so doesn't need to be restartable because it is stateless. When you have a stateful writer you should probably also be sure to implement ItemStream as well as ItemWriter. Remember also that the client of the writer needs to be aware of the ItemStream, so you may need to register it with a factory bean (e.g. one of the StepFactoryBean implementations in Spring Batch Core).

Because the script launching the job must kick off a Java Virtual Machine, there needs to be a class with a main method to act as the primary entry point. Spring Batch provides an implementation that serves just this purpose: CommandLineJobRunner. It's important to note that this is just one way to bootstrap your application, but there are many ways to launch a Java process, and this class should in no way be viewed as definitive. It performs four tasks:

All of these tasks are accomplished based completely upon the arguments passed in. The following are required arguments:

These arguments must be passed in with the path first and the name second. All arguments after these are considered to be JobParameters and must be in the format of 'name=value':

bash$ java CommandLineJobRunner endOfDayJob.xml endOfDay schedule.date(date)=2008/01/01

In most cases you would want to use a manifest to declare your main class in a jar, but for simplicity, the class was used directly. This example is using the same 'EndOfDay' example from Chapter 2. The first argument is 'endOfDayJob.xml', which is the Spring ApplicationContext containing the Job. The second argument, 'endOfDay' represents the job name. The final argument, 'schedule.date=01-01-2008' will be converted into JobParameters. An example of the XML configuration is below:

  <bean id="endOfDay"
        class="org.springframework.batch.core.job.SimpleJob">
    <property name="steps">
      <bean id="step1" parent="simpleStep" />
      <!-- Step details removed for clarity -->
    </property> 
  </bean>

  <!-- Launcher details removed for clarity -->
  <bean id="jobLauncher"
        class="org.springframework.batch.core.launch.support.SimpleJobLauncher" />

This example is overly simplistic, since there are many more requirements to a run a batch job in Spring Batch in general, but it serves to show the two main requirements of the CommandLineJobRunner: Job and JobLauncher

When launching a batch job from the command-line, it is often from an enterprise scheduler. Most schedulers are fairly dumb, and work only at the process level. Meaning, they only know about some operating system process such as a shell script that they're invoking. In this scenario, the only way to communicate back to the scheduler about the success or failure of a job is through return codes. A number is returned to a scheduler that is told how to interpret the result. In the simple case: 0 is success and 1 is failure. However, there may be scenarios such as: If job A returns 4 kick off job B, if it returns 5 kick off job C. This type of behavior is configured at the scheduler level, but it is important that a processing framework such as Spring Batch provide a way to return a numeric representation of of the 'Exit Code' for a particular batch job. In Spring Batch this is encapsulated within an ExitStatus, which is covered in more detail in Chapter 5. For the purposes of discussing exit codes, the only important thing to know is that an ExitStatus has an exit code property that is set by the framework (or the developer) and is returned as part of the JobExecution returned from the JobLauncher. The CommandLineJobRunner converts this string value to a number using the ExitCodeMapper interface:

  public interface ExitCodeMapper {

    public int intValue(String exitCode);
}

The essential contract of an ExitCodeMapper is that, given a string exit code, a number representation will be returned. The default implementation used by the job runner is the SimpleJvmExitCodeMapper that returns 0 for completion, 1 for generic errors, and 2 for any job runner errors such as not being able to find a Job in the provided context. If anything more complex than the 3 values above is needed, then a custom implementation of the ExitCodeMapper interface must be supplied. Because the CommandLineJobRunner is the class that creates an ApplicationContext, and thus cannot be 'wired together', any values that need to be overwritten must be autowired. This means that if an implementation of ExitCodeMapper is found within the BeanFactory, it will be injected into the runner after the context is created. All that needs to be done to provide your own ExitCodeMapper is to declare the implementation as a root level bean, and ensure it's part of the ApplicationContext that is loaded by the runner.

One of the most common reasons for wanting to launching a job asynchronously is to be able to gracefully stop it. This can be done through the JobExecution returned by the JobLauncher:

  JobExecution jobExecution = launcher.run(getJob(), jobParameters);

  //give job adequate time to start
  Thread.sleep(1000);

  assertEquals(BatchStatus.STARTED, jobExecution.getStatus());
  assertTrue(jobExecution.isRunning());

  jobExecution.stop();

  //give job time to stop
  Thread.sleep(1000);

  assertEquals(BatchStatus.STOPPED, jobExecution.getStatus());
  assertFalse(jobExecution.isRunning());

The shutdown is not immediate, since there is no way to force immediate shutdown, especially if the execution is currently in developer code that the framework has no control over, such as a business service. What it does mean, is that as soon as control is returned back to the framework, it will set the status of the current StepExecution to BatchStatus.STOPPED, save it, then do the same for the JobExecution before finishing.

Including the incrementers, which must be database specific, the configuration above is verbose. In order to make this more manageable, the framework provides a FactoryBean for convenience: JobRepositoryFactoryBean.

  <bean id="jobRepository"
        class="org.springframework.batch.execution.repository.JobRepositoryFactoryBean"
    <property name="databaseType" value="hsql" /> 
    <property name="dataSource" ref="dataSource" />
    <property name="transactionManager" ref="transactionManager" />
  </bean>

The databaseType property indicates the type of incrementer that must be used. Options include: "db2", "db2zos", "derby", "hsql", "mysql", "oracle", and "postgres".

There are scenarios in which you may not want to persist your domain objects to the database. One reason may be speed, storing domain objects at each commit point takes extra time. Another reason may be that you just don't need to persist status for a particular job. Spring batch provides a solution:

  <bean id="simpleJobRepository" class="org.springframework.batch.core.repository.support.SimpleJobRepository"> 
    <constructor-arg ref="mapJobInstanceDao" /> 
    <constructor-arg ref="mapJobExecutionDao" /> 
    <constructor-arg ref="mapStepExecutionDao" /> 
  </bean> 

  <bean id="mapJobInstanceDao" 
        class="org.springframework.batch.core.repository.dao.MapJobInstanceDao" /> 

  <bean id="mapJobExecutionDao" 
        class="org.springframework.batch.core.repository.dao.MapJobExecutionDao" /> 

  <bean id="mapStepExecutionDao" 
        class="org.springframework.batch.core.repository.dao.MapStepExecutionDao" /> 

The Map* DAO implementations store the batch artifacts in a transactional map. So, the repository and DAOs may still be used normally, and are transactionally sound, but their contents will be lost when the class is destroyed.

There is also a separate FactoryBean for the in-memory JobRepository, which reduces the amount of configuration required:

  <bean id="jobRepository" class="org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean" />

  <bean id="jobRepository"
        class="org.springframework.batch.execution.repository.JobRepositoryFactoryBean"
    <property name="databaseType" value="hsql" /> 
    <property name="dataSource" ref="dataSource" />
    <property name="transactionManager" ref="transactionManager" />
    <property name="IsolationLevelForCreate" value="ISOLATION_REPEATABLE_READ" />
  </bean>

  
  <aop:config>
      <aop:advisor 
           pointcut="execution(* org.springframework.batch.core..*Repository+.*(..))"
      <advice-ref="txAdvice" />
  </aop:config>

  <tx:advice id="txAdvice" transaction-manager="transactionManager">
      <tx:attributes>
          <tx:method name="*" />
      </tx:attributes>
  </tx:advice>

One key concern when execution a batch job, is what happens when a failed job is restarted? A Job is considered to have been 'restarted' if the same JobInstance has more than one JobExecution. Ideally, all jobs should be able to start up where they left off, but there are scenarios where this is not possible. It is entirely up to the developer to ensure that a new instance is always created in this scenario. However, Spring Batch does provide some help. If a Job should never be restarted, but should always be run as part of a new JobInstance, then the restartable property may be set to 'false':

  <bean id="footballJob"
        class="org.springframework.batch.core.job.SimpleJob">
    <property name="steps">
      <list>
        <!-- Step Bean details ommitted for clarity -->
        <bean id="playerload" parent="simpleStep" />
        <bean id="gameLoad" parent="simpleStep" />
        <bean id="playerSummarization" parent="simpleStep" />
      </list>
    </property>
    <property name="jobRepository" ref="jobRepository" />
    <property name="restartable" value="false" />
  </bean>

To phrase it another way, setting restartable to false means "this Job does not support being started again". Restarting a Job that is not restartable will cause a JobRestartException to be thrown:

  Job job = new SimpleJob();
  job.setRestartable(false);

  JobParameters jobParameters = new JobParameters();

  JobExecution firstExecution = jobRepository.createJobExecution(job, jobParameters);
  jobRepository.saveOrUpdate(firstExecution);

  try {
    jobRepository.createJobExecution(job, jobParameters);
    fail();
  }
  catch (JobRestartException e) {
    // expected
  }

This snippet of JUnit code shows how attempting to create a JobExecution the first time for a non restartable job will cause no issues. However, the second attempt will throw a JobRestartException.

During the course of the execution of a Job, it may be useful to be notified of various events in its lifecycle so that custom code may be executed. The SimpleJob allows for this by calling a JobListener at the appropriate time:

  public interface JobListener {

    void beforeJob(JobExecution jobExecution);

    void afterJob(JobExecution jobExecution);

    void onError(JobExecution jobExecution, Throwable e);

    void onInterrupt(JobExecution jobExecution);
  }

Listeners can be added to a SimpleJob via the setJobListeners property:

  <bean id="footballJob"
        class="org.springframework.batch.core.job.SimpleJob">
    <property name="steps">
      <list>
        <!-- Step Bean details ommitted for clarity -->
        <bean id="playerload" parent="simpleStep" />
        <bean id="gameLoad" parent="simpleStep" />
        <bean id="playerSummarization" parent="simpleStep" />
      </list>
    </property>
    <property name="jobRepository" ref="jobRepository" />
  <property name="jobListeners">
    <bean class="org.springframework.batch.core.listener.JobListenerSupport" />
  </property>
  </bean>

Despite the relatively short list of required dependencies for an ItemOrientedStep, it is an extremely complex class that can potentially contain many collaborators. In order to ease configuration, a SimpleStepFactoryBean can be used:

  <bean id="simpleStep"
        class="org.springframework.batch.core.step.item.SimpleStepFactoryBean" >
    <property name="transactionManager" ref="transactionManager" />
    <property name="jobRepository" ref="jobRepository" />
    <property name="itemReader" ref="itemReader" />
    <property name="itemWriter" ref="itemWriter" />
  </bean>

The configuration above represents the only required dependencies of the factory bean. Attempting to instantiate a SimpleStepFactoryBean without at least those four dependencies will result in an exception being thrown during construction by the Spring container.

As mentioned above, the ItemOrientedStep reads in and writes out items, periodically commiting using the supplied PlatformTransactionManager. By default, it will commit after each item has been written. This is less than ideal in many situations, since beginning and commiting a transaction is expensive. Ideally, you would like to process as many items as possible in each transaction, which is completely dependant upon the type of data being processed and the resources that are being interacted with. For this reason, the number of items that are processed within a commit can be set as the commit interval:

 <bean id="simpleStep"
        class="org.springframework.batch.core.step.item.SimpleStepFactoryBean" >
    <property name="transactionManager" ref="transactionManager" />
    <property name="jobRepository" ref="jobRepository" />
    <property name="itemReader" ref="itemReader" />
    <property name="itemWriter" ref="itemWriter" />
    <property name="commitInterval" value="10" />
  </bean>

In this example, 10 items will be processed within each transaction. At the beginning of processing a transaction is begun, and each time read is called on the ItemReader, a counter is incremented. When it reaches 10, the transaction will be committed.

Earlier in this chapter, restarting a Job was discussed. Restart has numerous impacts on steps, and as such may require some specific configuration.

 <bean id="simpleStep"
        class="org.springframework.batch.core.step.item.SimpleStepFactoryBean" >
    <property name="transactionManager" ref="transactionManager" />
    <property name="jobRepository" ref="jobRepository" />
    <property name="itemReader" ref="itemReader" />
    <property name="itemWriter" ref="itemWriter" />
    <property name="commitInterval" value="10" />
    <property name="startLimit" value="1" />
  </bean>

 <bean id="simpleStep"
        class="org.springframework.batch.core.step.item.SimpleStepFactoryBean" >
    <property name="transactionManager" ref="transactionManager" />
    <property name="jobRepository" ref="jobRepository" />
    <property name="itemReader" ref="itemReader" />
    <property name="itemWriter" ref="itemWriter" />
    <property name="commitInterval" value="10" />
    <property name="startLimit" value="1" />
    <property name="allowStartIfComplete" value="true" />
  </bean>

  <bean id="footballJob"
        class="org.springframework.batch.core.job.SimpleJob">
    <property name="steps">
      <list>
        <!-- Step Bean details ommitted for clarity -->
        <bean id="playerload" parent="simpleStep" />
        <bean id="gameLoad" parent="simpleStep" >
          <property name="allowStartIfComplete" value="true" />
        </bean>
        <bean id="playerSummarization" parent="simpleStep" >
          <property name="startLimit" value="2" />
        </bean>
      </list>
    </property>
    <property name="jobRepository" ref="jobRepository" />
    <property name="restartable" value="true" />
  </bean>

There are many scenarios where errors encountered while processing should not result in Step failure, but should be skipped instead. This is usually a decision that must be made by someone who understands the data itself and what meaning it has. Financial data, for example, may not be skippable because it results in money being transferred, which needs to be completely accurate. Loading in a list of vendors, on the other hand, might allow for skips, since a vendor not being loaded because it was formatted incorrectly, or missing necessary information, won't cause issues. Usually these bad records are logged as well, which will be covered later when discussing listeners. Configuring skip handling requires using a new factory bean: SkipLimitStepFactoryBean

  <bean id="skipSample"
        class="org.springframework.batch.core.step.item.SkipLimitStepFactoryBean">
    <property name="skipLimit" value="10" />
    <property name="itemReader" ref="flatFileItemReader" />
    <property name="itemWriter" ref="itemWriter" />
    <property name="skippableExceptionClasses"
              value="org.springframework.batch.item.file.FlatFileParseException">
    </property>
  </bean> 

In this example, a FlatFileItemReader is used, and if at any point a FlatFileParseException is thrown, it will be skipped and counted against the total skip limit of 10. It should be noted that any failures encountered while reading will not count against the commit interval. In other words, the commit interval is only incremented on writes (regardless of success or failure).

One problem with the example above is that any other exception besides a FlatFileParseException will cause the Job to fail. In certain scenarios this may be the correct behaviour, however, in certain scenarios it may be easier to identify which exceptions should cause failure and skip everything else:

  <bean id="skipSample"
        class="org.springframework.batch.core.step.item.SkipLimitStepFactoryBean">
    <property name="skipLimit" value="10" />
    <property name="itemReader" ref="flatFileItemReader" />
    <property name="itemWriter" ref="itemWriter" />
    <property name="skippableExceptionClasses"
              value="java.lang.Exception">
    <property name="fatalExceptionClasses"
              value="java.io.FileNotFoundException">
    </property>
  </befan>

By setting the skippable exceptions to java.lang.Exception, any exception that is thrown will be skipped. However, the second list, 'fatalExceptionClasses', contains specific exceptions that should be fatal if encountered.

In most cases you want an Exception to cause either a skip or Step failure. However, not all exceptions are deterministic. If a FlatFileParseException is encountered while reading, it will always be thrown for that record. Resseting the ItemReader will not help. However, for other exceptions, such as a DeadlockLoserDataAccessException, which indicates that the current process has attempted to update a record that another process holds a lock on, waiting and trying again might result in success. In this case, retry should be configured:

  <bean id="step1"
        class="org.springframework.batch.core.step.item.SkipLimitStepFactoryBean">
    <property name="itemReader" ref="itemGenerator" />
    <property name="itemWriter" ref="itemWriter" />
    <property name="retryLimit" value="3" />
    <property name="retryableExceptionClasses" value="org.springframework.dao.DeadlockLoserDataAccessException" />
  </bean>

The SkipLimitStepFactoryBean requires a limit for the number of times an individual item can be retried, and a list of Exceptions that are 'retryable'.

By default, regardless of retry or skip, any exceptions thrown from the ItemWriter will cause the transaction controlled by the Step to rollback. If skip is configured as described above, exceptions thrown from the ItemReader will not cause a rollback. However, there are many scenarios in which exceptions thrown from the ItemWriter should not cause a rollback because no action has taken place to invalidate the transaction. For this reason, the SkipLimitStepFactoryBean can be configured with a list of exceptions that should not cause rollback:

  <bean id="step2" 
        class="org.springframework.batch.core.step.item.SkipLimitStepFactoryBean">
    <property name="commitInterval" value="2" />
    <property name="skipLimit" value="1" />
    <!-- No rollback for exceptions that are marked with "+" in the tx attributes -->
    <property name="transactionAttribute"
      value="+org.springframework.batch.item.validator.ValidationException" />
    <property name="itemReader"
              ref="tradeSqlItemReader" />
    <property name="itemWriter"
              ref="itemTrackingWriter" />
  </bean>

The TransactionAttribute property above can be used to control multiple other settings such as isolation and propagation behaviour. More information on setting transaction attributes can be found in the spring core documentation.

The step has to take care of ItemStream callbacks at the necessary points in its lifecycle. This is vital if a step fails, and might need to be restarted, because the ItemStream interface is where the step gets the information it needs about persistent state between executions. The factory beans that Spring Batch provides for convenient configuration of Step instances have features that allow streams to be registered with the step when it is configured.

If the ItemReader or ItemWriter themselves implement the ItemStream interface, then these will be registered automatically. Any other streams need to be registered separately. This is often the case where there are indirect dependencies, like delegates being injected into the reader and writer. To register these they can be injected into the factory beans through the streams property, as illustrated below:

<bean id="step1"
        class="org.springframework.batch.core.step.item.SkipLimitStepFactoryBean">
    <property name="streams" ref="fileItemReader" />
    <property name="itemReader">
        <bean
            class="org.springframework.batch.item.validator.ValidatingItemReader">
            <property name="itemReader" ref="itemReader" />
            <property name="validator" ref="fixedValidator" />
        </bean>
    </property>
    ...
</bean>

In the example above the main item reader is being set up to delegate to a bean called "fileItemReader", which itself is being registered as a stream directly. The step will now be restartable and the state of the reader will be correctly persisted in case of a failure.

Just as with the Job, there are many events during the execution of a Step that a user may need notification of. For example, if writing out to a flat file that requires a footer, the ItemWriter needs to be notified when the Step has been completed, so that it can write the footer. This can be accomplished with one of many Step scoped listeners.

  public interface StepExecutionListener extends StepListener {

    void beforeStep(StepExecution stepExecution);

    ExitStatus onErrorInStep(StepExecution stepExecution, Throwable e);

    ExitStatus afterStep(StepExecution stepExecution);
}

  <bean id="simpleStep"
        class="org.springframework.batch.core.step.item.SimpleStepFactoryBean" >
    <property name="transactionManager" ref="transactionManager" />
    <property name="jobRepository" ref="jobRepository" />
    <property name="itemReader" ref="itemReader" />
    <property name="itemWriter" ref="itemWriter" />
    <property name="commitInterval" value="10" />
    <property name="listeners" ref="stepListener" />    
  </bean>

  public interface ChunkListener extends StepListener {

    void beforeChunk();

    void afterChunk();
  }

  public interface ItemReadListener extends StepListener {
  
    void beforeRead();

    void afterRead(Object item);
    
    void onReadError(Exception ex);
}

  public interface ItemWriteListener extends StepListener {

    void beforeWrite(Object item);

    void afterWrite(Object item);

    void onWriteError(Exception ex, Object item);
}

  public interface SkipListener extends StepListener {

    void onSkipInRead(Throwable t);

    void onSkipInWrite(Object item, Throwable t);
  }

As with other adapters for the ItemWriter and ItemReader interfaces, the Tasklet interface contains an implementation that allows for adapting itself to any pre-existing class: TaskletAdapter. An example where this may be useful is an existing DAO that is used to upate a flag on a set of records. The TaskletAdapter can be used to call this class without having to write an adapter for the Tasklet interface:

  <bean id="deleteFilesInDir" parent="taskletStep">
    <property name="tasklet">
      <bean class="org.springframework.batch.core.step.tasklet.TaskletAdapter">
        <property name="targetObject">
          <bean class="org.mycompany.FooDao">
        </property>
        <property name="targetMethod" value-"updateFoo" />
      </bean>
    </property>
  </bean>

Many batch jobs contains steps that must be done before the main processing begins in order to set up various resources, or after processing has completed to cleanup those resources. In the case of a job that works heavily with files, it is often necessary to delete certain files locally after they have been uploaded successfully to another location. The example below taken from the Spring Batch samples project, is a Tasklet implementation with just such a responsibility:

  public class FileDeletingTasklet implements Tasklet, InitializingBean {

    private Resource directory;

    public ExitStatus execute() throws Exception {
      File dir = directory.getFile();
      Assert.state(dir.isDirectory());
    
      File[] files = dir.listFiles();
      for (int i = 0; i < files.length; i++) {
        boolean deleted = files[i].delete();
        if (!deleted) {
          throw new UnexpectedJobExecutionException("Could not delete file " + files[i].getPath());
        }
      }
      return ExitStatus.FINISHED;
    }

    public void setDirectoryResource(Resource directory) {
      this.directory = directory;
    }

    public void afterPropertiesSet() throws Exception {
      Assert.notNull(directory, "directory must be set");
    }
  }

The above Tasklet implementation will delete all files within a given directory. It should be noted that the execute method will only be called once. All that is left is to inject the Tasklet into a TaskletStep:

  <bean id="taskletJob" parent="simpleJob">
    <property name="steps">
      <bean id="deleteFilesInDir" parent="taskletStep">
        <property name="tasklet">
          <bean class="org.springframework.batch.sample.tasklet.FileDeletingTasklet">
            <property name="directoryResource" ref="directory" />
          </bean>
        </property>
      </bean>
    </property>
  </bean>

  <bean id="directory"
        class="org.springframework.core.io.FileSystemResource">
    <constructor-arg value="target/test-outputs/test-dir" />
  </bean>

Many batch jobs may require that an external command be called from within the batch job. Such a process could be kicked off separately by the scheduler, but the advantage of common meta-data about the run would be lost. Furthermore, a multi-step job would also need to be split up into multiple jobs as well. Because the need is so common, Spring Batch provides a Tasklet implementation for calling system commands:

  <bean class="org.springframework.batch.sample.tasklet.SystemCommandTasklet">
    <property name="command" value="echo hello" />
    <!-- 5 second timeout for the command to complete -->
    <property name="timeout" value="5000" />
  </bean>

Part of the reponsibility of a stateful retry policy is to recognise the failed operations when they come back in a new transaction. To facilitate this in the commonest case where an object (like a message or message payload) is being processed, Spring Batch provides the ItemWriterRetryPolicy. This works in conjunction with a special RetryCallback implementation ItemWriterRetryCallback, which in turn relies on the user providing an ItemWriter. This callback implements the common pattern where it passes the item to a writer.

The way the failed operations are recognised in this implementation is by identifying the item across multiple invocations of the retry. To identify the item the user can provide an ItemKeyGenerator strategy, and this is responsible for returning a unique key identifying the item. The identifier is used as a key in the RetryContextCache. An ItemKeyGenerator can be provided either by injecting it directly into the ItemWriterRetryCallback, or by implementing the interface in the ItemWriter, or by accepting the default which is to simply use the item itself as a key.

When the retry is exhausted, because a stateful retry is always in a fresh transaction, there is also the option to handle the failed item in a different way, instead of calling the RetryCallback (which is presumed now to be likely to fail). This option is provided by the ItemRecoverer strategy. Like the key generator, it can be directly injected or provided by implementing the interface in the ItemWriter.

The decision to retry or not is actually delegated to a regular stateless retry policy, so the usual concerns about limits and timeouts can be injected into the ItemWriterRetryPolicy through the delegate property.