The Job Configuration

The configuration for the job is as follows:

package com.apress.springrecipes.springbatch.config;

import com.apress.springrecipes.springbatch.UserRegistration;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.item.ItemReader;                                                                                                              
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.database.BeanPropertyItemSqlParameterSourceProvider;
import org.springframework.batch.item.database.JdbcBatchItemWriter;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.LineMapper;
import org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper;
import org.springframework.batch.item.file.mapping.DefaultLineMapper;
import org.springframework.batch.item.file.transform.DelimitedLineTokenizer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.Resource;

import javax.sql.DataSource;

@Configuration                                                                          
public class UserJob {

    private static final String INSERT_REGISTRATION_QUERY =
        "insert into USER_REGISTRATION (FIRST_NAME, LAST_NAME, COMPANY, ADDRESS,CITY,STATE,ZIP,COUNTY,URL,PHONE_NUMBER,FAX)" +
        " values " +
        "(:firstName,:lastName,:company,:address,:city,:state,:zip,:county,:url,:phoneNumber,:fax)";

    @Autowired
    private JobBuilderFactory jobs;

    @Autowired
    private StepBuilderFactory steps;

    @Autowired
    private DataSource dataSource;

    @Value("file:${user.home}/batches/registrations.csv")
    private Resource input;                                                                                                              

    @Bean
    public Job insertIntoDbFromCsvJob() {
        return jobs.get("User Registration Import Job")
            .start(step1())
            .build();
    }

    @Bean                                                                          
    public Step step1() {
        return steps.get("User Registration CSV To DB Step")
            .<UserRegistration,UserRegistration>chunk(5)
            .reader(csvFileReader())
            .writer(jdbcItemWriter())
            .build();
    }

    @Bean
    public FlatFileItemReader<UserRegistration> csvFileReader() {
        FlatFileItemReader<UserRegistration> itemReader = new FlatFileItemReader<>();
        itemReader.setLineMapper(lineMapper());
        itemReader.setResource(input);
        return itemReader;
    }

    @Bean
    public JdbcBatchItemWriter<UserRegistration> jdbcItemWriter() {
        JdbcBatchItemWriter<UserRegistration> itemWriter = new JdbcBatchItemWriter<>();
        itemWriter.setDataSource(dataSource);
        itemWriter.setSql(INSERT_REGISTRATION_QUERY);
        itemWriter.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>());
        return itemWriter;
    }

    @Bean
    public DefaultLineMapper<UserRegistration> lineMapper() {
        DefaultLineMapper<UserRegistration> lineMapper = new DefaultLineMapper<>();
        lineMapper.setLineTokenizer(tokenizer());
        lineMapper.setFieldSetMapper(fieldSetMapper());
        return lineMapper;
    }

    @Bean
    public BeanWrapperFieldSetMapper<UserRegistration> fieldSetMapper() {
        BeanWrapperFieldSetMapper<UserRegistration> fieldSetMapper = new BeanWrapperFieldSetMapper<>();                                                                                                              
        fieldSetMapper.setTargetType(UserRegistration.class);
        return fieldSetMapper;
    }

    @Bean
    public DelimitedLineTokenizer tokenizer() {
        DelimitedLineTokenizer tokenizer = new DelimitedLineTokenizer();
        tokenizer.setDelimiter(",");
        tokenizer.setNames(new String[]{"firstName","lastName","company","address","city","state","zip","county","url","phoneNumber","fax"});
        return tokenizer;
    }
}

As described earlier, a job consists of steps, which are the real workhorse of a given job. The steps can be as complex or as simple as you like. Indeed, a step could be considered the smallest unit of work for a job. Input (what’s read) is passed to the step and potentially processed; then output (what’s written) is created from the step. This processing is spelled out using a Tasklet (which is another interface provided by Spring Batch). You can provide your own Tasklet implementation or simply use some of the preconfigured configurations for different processing scenarios. These implementations are made available in terms of configuration methods. One of the most important aspects of batch processing is chunk-oriented processing, which is employed here using the chunk() configuration method.

In chunk-oriented processing, input is read from a reader, optionally processed, and then aggregated. Finally, at a configurable interval—as specified by the commit-interval attribute to configure how many items will be processed before the transaction is committed—all the input is sent to the writer. If there is a transaction manager in play, the transaction is also committed. Right before a commit, the metadata in the database is updated to mark the progress of the job.

There are some nuances surrounding the aggregation of the input (read) values when a transaction-aware writer (or processor) rolls back. Spring Batch caches the values it reads and writes them to the writer. If the writer component is transactional, like a database, and the reader is not, there’s nothing inherently wrong with caching the read values and perhaps retrying or taking some alternative approach. If the reader itself is also transactional, then the values read from the resource will be rolled back and could conceivably change, rendering the in-memory cached values stale. If this happens, you can configure the chunk to not cache the values using reader-transactional-queue="true" on the chunk element .

Input

The first responsibility is reading a file from the file system. You use a provided implementation for the example. Reading CSV files is a common scenario, and Spring Batch’s support does not disappoint. The org.springframework.batch.item.file.FlatFileItemReader<T> class delegates the task of delimiting fields and records within a file to a LineMapper<T>, which in turn delegates the task of identifying the fields within that record to LineTokenizer. You use an org.springframework.batch.item.file.transform.DelimitedLineTokenizer, which is configured to delineate fields separated by a comma (,) character.

The DefaultLineMapper also declares a fieldSetMapper attribute that requires an implementation of FieldSetMapper. This bean is responsible for taking the input name-value pairs and producing a type that will be given to the writer component.

In this case, you use a BeanWrapperFieldSetMapper that will create a JavaBean POJO of type UserRegistration. You name the fields so that you can reference them later in the configuration. These names don’t have to be the values of some header row in the input file; they just have to correspond to the order in which the fields are found in the input file. These names are also used by the FieldSetMapper to match properties on a POJO. As each record is read, the values are applied to an instance of a POJO, and that POJO is returned.

@Bean                                                                          
public FlatFileItemReader<UserRegistration> csvFileReader() {
    FlatFileItemReader<UserRegistration> itemReader = new FlatFileItemReader<>();
    itemReader.setLineMapper(lineMapper());
    itemReader.setResource(input);
    return itemReader;
}

@Bean
public DefaultLineMapper<UserRegistration> lineMapper() {
    DefaultLineMapper<UserRegistration> lineMapper = new DefaultLineMapper<>();
    lineMapper.setLineTokenizer(tokenizer());
    lineMapper.setFieldSetMapper(fieldSetMapper());
    return lineMapper;
}

@Bean
public BeanWrapperFieldSetMapper<UserRegistration> fieldSetMapper() {
    BeanWrapperFieldSetMapper<UserRegistration> fieldSetMapper = new BeanWrapperFieldSetMapper<>();
    fieldSetMapper.setTargetType(UserRegistration.class);
    return fieldSetMapper;                                                                                                              
}

@Bean
public DelimitedLineTokenizer tokenizer() {
    DelimitedLineTokenizer tokenizer = new DelimitedLineTokenizer();
    tokenizer.setDelimiter(",");
    tokenizer.setNames(new String[]{"firstName","lastName","company","address","city","state","zip","county","url","phoneNumber","fax"});
    return tokenizer;
}

The class returned from the reader, UserRegistration, is a rather plain JavaBean .

package com.apress.springrecipes.springbatch;

public class UserRegistration implements Serializable {

    private String firstName;
    private String lastName;
    private String company;
    private String address;
    private String city;
    private String state;
    private String zip;
    private String county;                                                                                                              
    private String url;
    private String phoneNumber;
    private String fax;

   //... accessor / mutators omitted for brevity ...                

}

Output

The next component to do work is the writer, which is responsible for taking the aggregated collection of items read from the reader. In this case, you might imagine that a new collection (java.util.List<UserRegistration>) is created, then written, and then reset each time the collection exceeds the commit-interval attribute on the chunk element. Because you’re trying to write to a database, you use Spring Batch’s org.springframework.batch.item.database.JdbcBatchItemWriter. This class contains support for taking input and writing it to a database. It is up to the developer to provide the input and to specify what SQL should be run for the input. It will run the SQL specified by the sql property, in essence reading from the database, as many times as specified by the chunk element’s commit-interval and then commit the whole transaction. Here, you’re doing a simple insert. The names and values for the named parameters are being created by the bean configured for the itemSqlParameterSourceProvider property, which is an instance of BeanPropertyItemSqlParameterSourceProvider, whose sole job it is to take JavaBean properties and make them available as named parameters corresponding to the property name on the JavaBean.

@Bean
public JdbcBatchItemWriter<UserRegistration> jdbcItemWriter() {
    JdbcBatchItemWriter<UserRegistration> itemWriter = new JdbcBatchItemWriter<>();
    itemWriter.setDataSource(dataSource);                                                                                                    
    itemWriter.setSql(INSERT_REGISTRATION_QUERY);
    itemWriter.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>());
    return itemWriter;
}

That’s it! A working solution. With little configuration and no custom code, you’ve built a solution for taking large CSV files and reading them into a database. This solution is bare-bones and leaves a lot of edge cases uncared for. You might want to do processing on the item as it’s read (before it’s inserted), for example.

This exemplifies a simple job. It’s important to remember that there are similar classes for doing the exact opposite transformation: reading from a database and writing to a CSV file.

@Bean
public Job insertIntoDbFromCsvJob() {
    return jobs.get("User Registration Import Job")
        .start(step1())
        .build();
}                                                                                                              

@Bean
public Step step1() {
    return steps.get("User Registration CSV To DB Step")
        .<UserRegistration,UserRegistration>chunk(5)
        .reader(csvFileReader())                                                                                                              
        .writer(jdbcItemWriter())
        .build();
}

To configure the step, you give it the name User Registration CSV To DB Step. You are using chunk-based processing, and you need to tell it that you want a chunk size of 5. Next you supply it with a reader and writer, and finally you tell the factory to build to the step. The configured step is finally wired as a starting point to your job, named User Registration Import Job, which consists only of this step.

Simplify the ItemReader and ItemWriter Configuration

Configuring the ItemReader and ItemWriter can be a daunting task. You need to know a quite a lot of the internals of Spring Batch (which classes to use, etc.). As of Spring Batch 4, configuring the readers and writers has become easier as there are now specific builders for the different readers and writers.

To configure the FlatFileItemReader, you could use the FlatFileItemReaderBuilder and instead of configuring four individual beans, it is now six lines of code (mainly because the formatting in the sample).

@Bean
public FlatFileItemReader<UserRegistration> csvFileReader() throws Exception {

    return new FlatFileItemReaderBuilder<UserRegistration>()
        .name(ClassUtils.getShortName(FlatFileItemReader.class))
        .resource(input)
        .targetType(UserRegistration.class)
        .delimited()
        .names(new String[]{"firstName","lastName","company","address","city","state","zip","county","url","phoneNumber","fax"})
        .build();
}

This builder will automatically create the DefaultLineMapper, BeanWrapperFieldSetMapper, and DelimitedLineTokenizer, and you don’t have to know that they are used internally. You can now basically describe your configuration rather than explicitly configuring all the different items.

The same can be applied to the JdbcBatchItemWriter using the JdbcBatchItemWriterBuilder.

@Bean
public JdbcBatchItemWriter<UserRegistration> jdbcItemWriter() {
    return new JdbcBatchItemWriterBuilder<UserRegistration>()
        .dataSource(dataSource)
        .sql(INSERT_REGISTRATION_QUERY)
        .beanMapped()
        .build();                                                                                                    
}

Write a Custom ItemReader

The ItemReader example is trivial. Here, an ItemReader is created that knows how to retrieve UserRegistration objects from a remote procedure call (RPC) endpoint:

package com.apress.springrecipes.springbatch;

import org.springframework.batch.item.ItemReader;

import java.util.Collection;
import java.util.Date;

public class UserRegistrationItemReader implements ItemReader<UserRegistration> {

    private final UserRegistrationService userRegistrationService;

    public UserRegistrationItemReader(UserRegistrationService userRegistrationService) {
        this.userRegistrationService = userRegistrationService;
    }

    public UserRegistration read() throws Exception {
        final Date today = new Date();
        Collection<UserRegistration> registrations = userRegistrationService.getOutstandingUserRegistrationBatchForDate(1, today);
        return registrations.stream().findFirst().orElse(null);
    }
}

As you can see, the interface is trivial. In this case, you defer most work to a remote service to provide you with the input. The interface requires that you return one record. The interface is parameterized to the type of object (the “item”) to be returned. All the read items will be aggregated and then passed to the ItemWriter.

Write a Custom ItemWriter

The ItemWriter example is also trivial. Imagine wanting to write by invoking a remote service using any of the numerous options for remoting that Spring provides. The ItemWriter<T> interface is parameterized by the type of item you’re expecting to write. Here, you expect a UserRegistration object from the ItemReader<T>. The interface consists of one method, which expects a List of the class’s parameterized type. These are the objects read from ItemReader<T> and aggregated. If your commit-interval were ten, you might expect ten or fewer items in the List.

package com.apress.springrecipes.springbatch;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.item.ItemWriter;

import java.util.List;

public class UserRegistrationServiceItemWriter implements ItemWriter<UserRegistration> {

    private static final Logger logger = LoggerFactory.getLogger(UserRegistrationServiceItemWriter.class);

    private final UserRegistrationService userRegistrationService;

    public UserRegistrationServiceItemWriter(UserRegistrationService userRegistrationService) {
        this.userRegistrationService = userRegistrationService;
    }

    public void write(List<?extends UserRegistration> items) throws Exception {
        items.forEach(this::write);
    }

    private void write(UserRegistration userRegistration) {
        UserRegistration registeredUserRegistration = userRegistrationService.registerUser(userRegistration);
        logger.debug("Registered: {}", registeredUserRegistration);

    }
}

Here, you’ve wired in the service’s client interface. You simply loop through the UserRegistration objects and invoke the service, which in turn hands you back an identical instance of UserRegistration. If you remove the gratuitous spacing, curly brackets , and logging output, it becomes two lines of code to satisfy the requirement.

The interface for UserRegistrationService follows:

package com.apress.springrecipes.springbatch;

import java.util.Collection;
import java.util.Date;

public interface UserRegistrationService {

    Collection<UserRegistration> getOutstandingUserRegistrationBatchForDate(
        int quantity, Date date);

    UserRegistration registerUser(UserRegistration userRegistrationRegistration);

}

In this example, you have no particular implementation for the interface, as it is irrelevant: it could be any interface that Spring Batch doesn’t know about already.

Chain Processors Together

Sometimes you might want to add extra processing that isn’t congruous with the goals of the processor you’ve already set up. Spring Batch provides a convenience class, CompositeItemProcessor<I,O>, that forwards the output of the filter to the input of the successive filter. In this way, you can write many, singly focused ItemProcessor<I,O> s and then reuse them and chain them as necessary.

@Bean
public CompositeItemProcessor<Customer, Customer> compositeBankCustomerProcessor() {
    List<ItemProcessor<Customer, Customer>> delegates = Arrays.asList(creditScoreValidationProcessor(), salaryValidationProcessor(), customerEligibilityProcessor());
    CompositeItemProcessor<Customer, Customer> processor = new CompositeItemProcessor<>();                                                                                                    
    processor.setDelegates(delegates);
    return processor;
}

The example created a simple workflow. The first ItemProcessor<T> will take an input of whatever’s coming from the ItemReader<T> configured for this job, presumably a Customer object. It will check the credit score of the Customer and, if approved, forward the Customer to the salary and income validation processor. If everything checks out there, the Customer will be forwarded to the eligibility processor, where the system is checked for duplicates or any other invalid data. It will finally be forwarded to the writer to be added to the output. If at any point in the three processors the Customer fails a check, the executing ItemProcessor can simply return null and arrest processing .

Transactions

Spring’s core framework provides first-class support for transactions. You simply wire up a PlatformTransactionManager and give Spring Batch a reference, just as you would in any regular JdbcTemplate or HibernateTemplate solution. As you build your Spring Batch solutions, you’ll be given opportunities to control how steps behave in a transaction. You’ve already seen some of the support for transactions baked right in.

The configuration used in all these examples established a DriverManagerDataSource and a DataSourceTransactionManager bean. The PlatformTransactionManager and DataSource were then wired to the JobRepository, which was in turn wired to the JobLauncher, which you used to launch all jobs thus far. This enabled all the metadata your jobs created to be written to the database in a transactional way.

You might wonder why there is no explicit mention of the transaction manager when you configured the JdbcItemWriter with a reference to dataSource. The transaction manager reference can be specified, but in your solutions, it wasn’t required because Spring Batch will, by default, try to pluck the PlatformTransactionManager named transactionManager from the context and use it. If you want to explicitly configure this, you can specify the transactionManager property on the tasklet configuration method. A simple transaction manager for JDBC work might look like this:

@Bean
protected Step step1() {
    return steps.get("step1")
        .<UserRegistration,UserRegistration>chunk(5)
        .reader(csvFileReader())
        .processor(userRegistrationValidationItemProcessor())
        .writer(jdbcItemWriter())
        .transactionManager(new DataSourceTransactionManager(dataSource))
        .build();                                                                              
}

Items read from an ItemReader<T> are normally aggregated. If a commit on the ItemWriter<T> fails, the aggregated items are kept and then resubmitted. This process is efficient and works most of the time. One place where it breaks semantics is when reading from a transactional resource (like a JMS queue or database). Reads from a message queue can and should be rolled back if the transaction they participate in (in this case, the transaction for the writer) fails .

@Bean
protected Step step1() {
    return steps.get("step1")
        .<UserRegistration,UserRegistration>chunk(5)
        .reader(csvFileReader()).readerIsTransactionalQueue()
        .processor(userRegistrationValidationItemProcessor())
        .writer(jdbcItemWriter())
        .transactionManager(new DataSourceTransactionManager(dataSource))
        .build();
}

Rollbacks

Handling the simple case (“read X items, and every Y items, commit a database transaction every Y items”) is easy. Spring Batch excels in the robustness it surfaces as simple configuration options for the edge and failure cases.

If a write fails on an ItemWriter, or some other exception occurs in processing, Spring Batch will roll back the transaction. This is valid handling for a majority of the cases. There may be some scenarios when you want to control which exceptional cases cause the transaction to roll back.

When using Java-based configuration to enable rollbacks, the first step needs to be a fault-tolerant step, which in turn can be used to specify the no-rollback exceptions. First use faultTolerant() to obtain a fault-tolerant step, next the skipLimit() method can be used to specify the number of ignored rollbacks before actually stopping the job execution, and finally the noRollback() method can be used to specify the exceptions that don’t trigger a rollback. To specify multiple exceptions, you can simply chain calls to the noRollback() method.

@Bean
protected Step step1() {
    return steps.get("step1")                                                                              
        .<UserRegistration,UserRegistration>chunk(10)
            .faultTolerant()
                .noRollback(com.yourdomain.exceptions.YourBusinessException.class)
        .reader(csvFileReader())
        .processor(userRegistrationValidationItemProcessor())
        .writer(jdbcItemWriter())
        .build();
}

Configure a Step

When using Java-based configuration to enable retrying, the first step needs to be a fault-tolerant step, which in turn can be used to specify the retry limit and retryable exceptions. First use faultTolerant() to obtain a fault-tolerant step, next the retryLimit() method can be used to specify the number of retry attempts, and finally the retry() method can be used to specify the exceptions that trigger a retry. To specify multiple exceptions, you can simply chain calls to the retry() method.

@Bean
public Step step1() {
    return steps.get("User Registration CSV To DB Step")                                                                                                    
        .<UserRegistration,UserRegistration>chunk(10)
            .faultTolerant()
                .retryLimit(3).retry(DeadlockLoserDataAccessException.class)
        .reader(csvFileReader())
        .writer(jdbcItemWriter())
        .transactionManager(transactionManager)
        .build();
}

Retry Template

Alternatively, you can leverage Spring Retry support for retries and recovery in your own code. For example, you can have a custom ItemWriter<T> in which retry functionality is desired or even an entire service interface for which retry support is desired.

Spring Batch supports these scenarios through the RetryTemplate that (much like its various other Template cousins) isolates your logic from the nuances of retries and instead enables you to write the code as though you were going to attempt it only once. Let Spring Batch handle everything else through declarative configuration.

The RetryTemplate supports many use cases, with convenient APIs to wrap otherwise tedious retry/fail/recover cycles in concise, single-method invocations.

Let’s take a look at the modified version of a simple ItemWriter<T> from recipe 11-4 on how to write a custom ItemWriter<T>. The solution was simple enough and would ideally work all the time. It fails to handle the error cases for the service, however. When dealing with RPC, always proceed as if it’s almost impossible for things to go right; the service itself may surface a semantic or system violation. An example might be a duplicate database key, invalid credit card number, and so on. This is true whether the service is distributed or in-VM, of course .

Next, the RPC layer below the system may also fault. Here’s the rewritten code, this time allowing for retries:

ppackage com.apress.springrecipes.springbatch;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.item.ItemWriter;
import org.springframework.retry.RetryCallback;
import org.springframework.retry.support.RetryTemplate;

import java.util.List;

public class RetryableUserRegistrationServiceItemWriter implements ItemWriter<UserRegistration> {

    private static final Logger logger = LoggerFactory.getLogger(RetryableUserRegistrationServiceItemWriter.class);

    private final UserRegistrationService userRegistrationService;
    private final RetryTemplate retryTemplate;                                                                          

    public RetryableUserRegistrationServiceItemWriter(UserRegistrationService userRegistrationService, RetryTemplate retryTemplate) {
        this.userRegistrationService = userRegistrationService;
        this.retryTemplate = retryTemplate;
    }

    public void write(List<?extends UserRegistration> items)
        throws Exception {
        for (final UserRegistration userRegistration : items) {
            UserRegistration registeredUserRegistration = retryTemplate.execute(
                (RetryCallback<UserRegistration, Exception>) context -> userRegistrationService.registerUser(userRegistration));

            logger.debug("Registered: {}", registeredUserRegistration);
        }
    }
}

As you can see, the code hasn’t changed much, and the result is much more robust . The RetryTemplate itself is configured in the Spring context, although it’s trivial to create in code. I declare it in the Spring context only because there is some surface area for configuration when creating the object, and I try to let Spring handle the configuration.

One of the more useful settings for the RetryTemplate is the BackOffPolicy in use. The BackOffPolicy dictates how long the RetryTemplate should back off between retries. Indeed, there’s even support for growing the delay between retries after each failed attempt to avoid lock stepping with other clients attempting the same invocation. This is great for situations in which there are potentially many concurrent attempts on the same resource and a race condition may ensue. There are other BackOffPolicy settings, including one that delays retries by a fixed amount called FixedBackOffPolicy.

@Bean
public RetryTemplate retryTemplate() {
    RetryTemplate retryTemplate = new RetryTemplate();
    retryTemplate.setBackOffPolicy(backOffPolicy());
    return retryTemplate;
}

@Bean
public ExponentialBackOffPolicy backOffPolicy() {
    ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
    backOffPolicy.setInitialInterval(1000);
    backOffPolicy.setMaxInterval(10000);
    backOffPolicy.setMultiplier(2);
    return backOffPolicy;
}

You have configured a RetryTemplate’s backOffPolicy so that backOffPolicy will wait 1 second (1,000 milliseconds) before the initial retry. Subsequent attempts will double that value (the growth is influenced by the multiplier). It’ll continue until the maxInterval is met, at which point all subsequent retry intervals will level off, retrying at a consistent interval.

AOP-Based Retries

An alternative is an AOP adviser provided by Spring Batch that will wrap invocations of methods whose success is not guaranteed in retries, as you did with the RetryTemplate. In the previous example, you rewrote an ItemWriter<T> to make use of the template. Another approach might be to merely advise the entire userRegistrationService proxy with this retry logic. In this case, the code could go back to the way it was in the original example, with no RetryTemplate!

To do so, you would annotate the method (or methods) to be retryable with the @Retryable annotation. To achieve the same as in the code with an explicit RetryTemplate, you would need to add the following.

@Retryable(backoff = @Backoff(delay = 1000, maxDelay = 10000, multiplier = 2))
public UserRegistration registerUser(UserRegistration userRegistrationRegistration) { ... }

Only adding this annotation isn’t enough; you would also need to enable annotation processing for this with the @EnableRetry annotation on your configuration .

@Configuration
@EnableBatchProcessing
@EnableRetry
@ComponentScan("com.apress.springrecipes.springbatch")
@PropertySource("classpath:/batch.properties")
public class BatchConfiguration { ... }

Sequential Steps

In the previous example, there’s an implied sequence between the first two steps; the audit file can’t be written until all the registrations have completed. This sort of relationship is the default relationship between two steps. One occurs after the other. Each step executes with its own execution context and shares only a parent job execution context and an order.

@Bean
public Job nightlyRegistrationsJob () {
    return jobs.get("nightlyRegistrationsJob ")
        .start(loadRegistrations())
        .next(reportStatistics())
        .next(...)
        .build();                                                                                                    
    }
}

Concurrency

The first version of Spring Batch was oriented toward batch processing inside the same thread and, with some alteration, perhaps inside the virtual machine. There were workarounds, of course, but the situation was less than ideal.

In the outline for this example job, the first step had to come before the second two because the second two are dependent on the first. The second two, however, do not share any such dependencies. There’s no reason why the audit log couldn’t be written at the same time as the JMS messages are being delivered. Spring Batch provides the capability to fork processing to enable just this sort of arrangement.

@Bean
public Job insertIntoDbFromCsvJob() {
    JobBuilder builder = jobs.get("insertIntoDbFromCsvJob");
    return builder
        .start(loadRegistrations())
        .split(taskExecutor())
            .add(
                builder.flow(reportStatistics()),
                builder.flow(sendJmsNotifications()))
        .build();
}

You can use the split() method on the job builder. To make a step into a flow , the flow() method of the job builder can be used; then, to add more steps to the flow, these can be added with the next() method. The split() method requires a TaskExecutor to be set; see recipe 2-23 for more information on scheduling and concurrency.

In this example, there’s nothing to prevent you from having many steps within the flow elements, nor was there anything preventing you from having more steps after the split element. The split element, like the step elements, takes a next attribute as well.

Spring Batch provides a mechanism to offload processing to another process. This distribution requires some sort of durable, reliable connection. This is a perfect use of JMS because it’s rock-solid and transactional, fast, and reliable. Spring Batch support is modeled at a slightly higher level, on top of the Spring Integration abstractions for Spring Integration channels. This support is not in the main Spring Batch code; it can be found in the spring-batch-integration project. Remote chunking lets individual steps read and aggregate items as usual in the main thread. This step is called the master. Items read are sent to the ItemProcessor<I,O>/ItemWriter<T> running in another process (this is called the slave). If the slave is an aggressive consumer, you have a simple, generic mechanism to scale: work is instantly farmed out over as many JMS clients as you can throw at it. The aggressive-consumer pattern refers to the arrangement of multiple JMS clients all consuming the same queue’s messages. If one client consumes a message and is busy processing, other idle queues will get the message instead. As long as there’s a client that’s idle, the message will be processed instantly.

Additionally, Spring Batch supports implicitly scaling out using a feature called partitioning. This feature is interesting because it’s built in and generally very flexible. You replace your instance of a step with a subclass, PartitionStep, which knows how to coordinate distributed executors and maintains the metadata for the execution of the step, thus eliminating the need for a durable medium of communication as in the “remote chunking” technology.

The functionality here is also very generic. It could, conceivably, be used with any sort of grid fabric technology such as GridGain or Hadoop. Spring Batch ships with only a TaskExecutorPartitionHandler, which executes steps in multiple threads using a TaskExecutor strategy. This simple improvement might be enough of a justification for this feature! If you’re really hurting, however, you can extend it .

Conditional Steps with Statuses

Using the ExitStatus of a given job or step to determine the next step is the simplest example of a conditional flow. Spring Batch facilitates this through the use of the stop, next, fail, and end elements. By default, assuming no intervention, a step will have an ExitStatus that matches its BatchStatus, which is a property whose values are defined in an enum and may be any of the following: COMPLETED, STARTING, STARTED, STOPPING, STOPPED, FAILED, ABANDONED, or UNKNOWN.

Let’s look at an example that executes one of two steps based on the success of a preceding step:

@Bean
public Job insertIntoDbFromCsvJob() {
    return jobs.get("User Registration Import Job")
        .start(step1())
            .on("COMPLETED").to(step2())
            .on("FAILED").to(failureStep())
        .build();
}

It’s also possible to provide a wildcard. This is useful if you want to ensure a certain behavior for any number of BatchStatus values, perhaps in tandem with a more specific next element that matches only one BatchStatus.

@Bean
public Job insertIntoDbFromCsvJob() {
    return jobs.get("User Registration Import Job")
        .start(step1())
            .on("COMPLETED").to(step2())
            .on("*").to(failureStep())
        .build();
}

In this example, you are instructing Spring Batch to perform some step based on any unaccounted-for ExitStatus. Another option is to just stop processing altogether with a BatchStatus of FAILED. You can do this using the fail element. A less aggressive rewrite of the preceding example might be the following:

@Bean
public Job insertIntoDbFromCsvJob() {
    return jobs.get("User Registration Import Job")
        .start(step1())
            .on("COMPLETED").to(step2())
            .on("FAILED").fail()
        .build();
}

In all these examples, you’re reacting to the standard BatchStatus values that the Spring Batch framework provides. But it’s also possible to raise your own ExitStatus. If, for example, you wanted the whole job to fail with a custom ExitStatus of MAN DOWN, you might do something like this:

@Bean
public Job insertIntoDbFromCsvJob() {
    return jobs.get("User Registration Import Job")
        .start(step1())
            .on("COMPLETED").to(step2())
            .on("FAILED").end("MAN DOWN")
        .build();
}

Finally, if all you want to do is end processing with a BatchStatus of COMPLETED, you can use the end() method. This is an explicit way of ending a flow as if it had run out of steps and incurred no errors .

@Bean
public Job insertIntoDbFromCsvJob() {
    return jobs.get("User Registration Import Job")
        .start(step1())
            .on("COMPLETED").end()
            .on("FAILED").to(errorStep())
        .build();
}

Conditional Steps with Decisions

If you want to vary the execution flow based on some logic more complex than a job’s ExitStatus values, you may give Spring Batch a helping hand by using a decision element and providing it with an implementation of a JobExecutionDecider.

package com.apress.springrecipes.springbatch;

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 HoroscopeDecider implements JobExecutionDecider {

    private boolean isMercuryIsInRetrograde () { return Math.random() > .9 ; }

    public FlowExecutionStatus decide(JobExecution jobExecution,
                                      StepExecution stepExecution) {
        if (isMercuryIsInRetrograde()) {
            return new FlowExecutionStatus("MERCURY_IN_RETROGRADE");                                                                                                              
        }
        return FlowExecutionStatus.COMPLETED;
    }
}

All that remains is the configuration, shown here:

@Bean                                                                  
public Job insertIntoDbFromCsvJob() {
    JobBuilder builder = jobs.get("insertIntoDbFromCsvJob");
    return builder
        .start(step1())
        .next((horoscopeDecider())
            .on("MERCURY_IN_RETROGRADE").to(step2())
            .on(("COMPLETED ").to(step3())
        .build();
}

Launch from a Web Application

Launching a job from a web application requires a slightly different approach because the client thread (presumably an HTTP request) can’t usually wait for a batch job to finish. The ideal solution is to have the job execute asynchronously when launched from a controller or action in the web tier, unattended by the client thread. Spring Batch supports this scenario through the use of a Spring TaskExecutor. This requires a simple change to the configuration for the JobLauncher, although the Java code can stay the same. Here, you will use a SimpleAsyncTaskExecutor that will spawn a thread of execution and manage that thread without blocking:

package com.apress.springrecipes.springbatch.config;

@Configuration
@EnableBatchProcessing
@ComponentScan("com.apress.springrecipes.springbatch")
@PropertySource("classpath:/batch.properties")
public class BatchConfiguration {

    @Bean
    public SimpleAsyncTaskExecutor taskExecutor() {
        return new SimpleAsyncTaskExecutor();
    }                                                                                                              
}

As you cannot use the default settings anymore, you need to add your own implementation of a BatchConfigurer to configure the TaskExecutor and add it to the SimpleJobLauncher. For this implementation, you used the DefaultBatchConfigurer as a reference; you only override the createJobLauncher method to add the TaskExecutor.

package com.apress.springrecipes.springbatch.config;

import org.springframework.batch.core.configuration.annotation.DefaultBatchConfigurer;                                                                                                              
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.launch.support.SimpleJobLauncher;
import org.springframework.core.task.TaskExecutor;
import org.springframework.stereotype.Component;

@Component
public class CustomBatchConfigurer extends DefaultBatchConfigurer {

    private final TaskExecutor taskExecutor;

    public CustomBatchConfigurer(TaskExecutor taskExecutor) {
        this.taskExecutor = taskExecutor;
    }

    @Override
    protected JobLauncher createJobLauncher() throws Exception {
        SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
        jobLauncher.setJobRepository(getJobRepository());
        jobLauncher.setTaskExecutor(this.taskExecutor);
        jobLauncher.afterPropertiesSet();                                                                                                              
        return jobLauncher;
    }
}

Run from the Command Line

Another common use case is deployment of a batch process from a system scheduler such as cron or autosys, or even Window’s event scheduler. Spring Batch provides a convenience class that takes as its parameters the name of the XML application context (that contains everything required to run a job) as well as the name of the job bean itself. Additional parameters may be provided and used to parameterize the job. These parameters must be in the form name=value. An example invocation of this class on the command line (on a Linux/Unix system), assuming that you set up the classpath, might look like this:

java CommandLineJobRunner jobs.xml hourlyReport date=`date +%m/%d/%Y time=date +%H`

The CommandLineJobRunner will even return system error codes (0 for success, 1 for failure, and 2 for an issue with loading the batch job) so that a shell (such as used by most system schedulers) can react or do something about the failure. More complicated return codes can be returned by creating and declaring a top-level bean that implements the interface ExitCodeMapper, in which you can specify a more useful translation of exit status messages to integer-based error codes that the shell will see on process exit.

Run on a Schedule

Spring has support for a scheduling framework (see also recipe 3-22). This framework lends itself perfectly to running Spring Batch. First, let’s modify the existing application context configuration to enable scheduling by using the @EnableScheduling annotation and by adding a ThreadPoolTaskScheduler.

package com.apress.springrecipes.springbatch.config;

@Configuration
@EnableBatchProcessing
@ComponentScan("com.apress.springrecipes.springbatch")
@PropertySource("classpath:/batch.properties")
@EnableScheduling                                                                          
@EnableAsync
public class BatchConfiguration {

    @Bean
    public ThreadPoolTaskScheduler taskScheduler() {
        ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
        taskScheduler.setThreadGroupName("batch-scheduler");
        taskScheduler.setPoolSize(10);
        return taskScheduler;
    }

}

These imports enable the simplest possible support for scheduling. The preceding annotations ensure that any bean under the package com.apress.springrecipes.springbatch will be configured and scheduled as required. The scheduler bean is as follows:

package com.apress.springrecipes.springbatch.scheduler;

import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.JobParametersBuilder;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.Date;                                                                                                              

@Component
public class JobScheduler {

    private final JobLauncher jobLauncher;
    private final Job job;

    public JobScheduler(JobLauncher jobLauncher, Job job) {
        this.jobLauncher = jobLauncher;
        this.job = job;
    }

    public void runRegistrationsJob(Date date) throws Throwable {
        System.out.println("Starting job at " + date.toString());

        JobParametersBuilder jobParametersBuilder = new JobParametersBuilder();
        jobParametersBuilder.addDate("date", date);
        jobParametersBuilder.addString("input.file", "registrations");                                                                                                              

        JobParameters jobParameters = jobParametersBuilder.toJobParameters();

        JobExecution jobExecution = jobLauncher.run(job, jobParameters);

        System.out.println("jobExecution finished, exit code: " + jobExecution.getExitStatus().getExitCode());
    }

    @Scheduled(fixedDelay = 1000 * 10)
    public void runRegistrationsJobOnASchedule() throws Throwable {
        runRegistrationsJob(new Date());
    }
}

There is nothing particularly novel here; it’s a good study of how the different components of the Spring Framework work well together. The bean is recognized and becomes part of the application context because of the @Component annotation, which you enabled with the @ComponentScan annotation in your configuration class. There’s only one Job in the UserJob class and only one JobLauncher, so you simply have those autowired into your bean. Finally, the logic for kicking off a batch run is inside the runRegistrationsJob(java.util.Date date) method. This method could be called from anywhere. Your only client for this functionality is the scheduled method runRegistrationsJobOnASchedule. The framework will invoke this method for you, according to the timeline dictated by the @Scheduled annotation.

There are other options for this sort of thing; traditionally in the Java and Spring world, this sort of problem would be a good fit for Quartz. It might still be, as the Spring scheduling support isn’t designed to be as extensible as Quartz. If you are in an environment requiring more traditional, ops-friendly scheduling tools, there are of course old standbys like cron, autosys, and BMC, too.

Launch a Job with Parameters

A job is a prototype of a JobInstance. Job Parameters are used to provide a way of identifying a unique run of a job (a JobInstance). These JobParameters allow you to give input to your batch process, just as you would with a method definition in Java. You’ve seen the JobParameters in previous examples but not in detail. The JobParameters object is created as you launch the job using the JobLauncher. To launch a job called dailySalesFigures, with the date for the job to work with, you would write something like this:

package com.apress.springrecipes.springbatch;

import com.apress.springrecipes.springbatch.config.BatchConfiguration;
import org.springframework.batch.core.*;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;

import java.util.Date;

public class Main {
    public static void main(String[] args) throws Throwable {

        ApplicationContext context =
            new AnnotationConfigApplicationContext(BatchConfiguration.class);

        JobLauncher jobLauncher = context.getBean(JobLauncher.class);
        Job job = context.getBean("dailySalesFigures", Job.class);                                                                                                              

        jobLauncher.run(job, new JobParametersBuilder()
            .addDate( "date", new Date() ).toJobParameters());
    }
}

Access JobParameters

Technically, you can get at JobParameters via any of the ExecutionContexts (step and job). Once you have it, you can access the parameters in a type-safe way by calling getLong(), getString(), and so on. A simple way to do this is to bind to the @BeforeStep event, save the StepExecution, and iterate over the parameters this way. From here, you can inspect the parameters and do anything you want with them. Let’s look at that in terms of the ItemProcessor<I,O> you wrote earlier.

// ...                    
private StepExecution stepExecution;

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

public UserRegistration process(UserRegistration input) throws Exception {

    Map<String, JobParameter> params =  stepExecution.getJobParameters().getParameters();

    for (String jobParameterKey : params.keySet()) {
        System.out.println(String.format("%s=%s", jobParameterKey,
    params.get(jobParameterKey).getValue().toString()));
    }                                                                                                              

    Date date = stepExecution.getJobParameters().getDate("date");
    // etc ...                
}

This turns out to be of limited value. The 80 percent case is that you’ll need to bind parameters from the job’s launch to the Spring beans in the application context. These parameters are available only at runtime, whereas the steps in the XML application context are configured at design time. This happens in many places. Previous examples demonstrated ItemWriters<T> and ItemReaders<T> with hard-coded paths. That works fine unless you want to parameterize the file name. This is hardly acceptable unless you plan on using a job just once!

The core Spring Framework features an enhanced expression language that Spring Batch uses to defer binding of the parameter until the correct time (or, in this case, until the bean is in the correct scope). Spring Batch has the “step” scope for just this purpose. Let’s take a look at how you’d rework the previous example to use a parameterized file name for the ItemReader’s resource:

@Bean
@StepScope
public ItemReader<UserRegistration> csvFileReader(@Value("file:${user.home}/batches/#{jobParameters['input.fileName']}.csv") Resource input) { ... }

All you did is scope the bean (the FlatFileItemReader<T>) to the life cycle of a step (at which point those JobParameters will resolve correctly) and then used the EL syntax to parameterize the path to work off of.