Spring Batch

Full Reference: See advanced.md for skip/retry configuration, partitioning, listeners, testing with JobLauncherTestUtils, composite writers, and async processing.

Deep Knowledge: Use mcp__documentation__fetch_docs with technology: spring-boot and topic: batch for comprehensive documentation.

Quick Start

@Configuration @EnableBatchProcessing public class BatchConfig {

@Bean
public Job importJob(JobRepository jobRepository, Step step1) {
    return new JobBuilder("importJob", jobRepository)
        .incrementer(new RunIdIncrementer())
        .start(step1)
        .build();
}

@Bean
public Step step1(JobRepository jobRepository,
                  PlatformTransactionManager transactionManager,
                  ItemReader<InputData> reader,
                  ItemProcessor<InputData, OutputData> processor,
                  ItemWriter<OutputData> writer) {
    return new StepBuilder("step1", jobRepository)
        .<InputData, OutputData>chunk(100, transactionManager)
        .reader(reader)
        .processor(processor)
        .writer(writer)
        .build();
}

}

Core Components

Job

@Bean public Job complexJob(JobRepository jobRepository, Step extractStep, Step transformStep, Step loadStep) { return new JobBuilder("etlJob", jobRepository) .incrementer(new RunIdIncrementer()) .validator(jobParametersValidator()) .listener(jobExecutionListener()) .start(extractStep) .next(transformStep) .next(loadStep) .build(); }

// Job with decision @Bean public Job conditionalJob(JobRepository jobRepository, Step step1, Step step2, Step errorStep, JobExecutionDecider decider) { return new JobBuilder("conditionalJob", jobRepository) .start(step1) .next(decider) .on("COMPLETED").to(step2) .from(decider).on("FAILED").to(errorStep) .end() .build(); }

Step - Chunk Processing

@Bean public Step chunkStep(JobRepository jobRepository, PlatformTransactionManager txManager) { return new StepBuilder("chunkStep", jobRepository) .<Person, Person>chunk(100, txManager) .reader(reader()) .processor(processor()) .writer(writer()) .faultTolerant() .skipLimit(10) .skip(ValidationException.class) .retryLimit(3) .retry(TransientException.class) .build(); }

// Tasklet Step (for simple operations) @Bean public Step taskletStep(JobRepository jobRepository, PlatformTransactionManager txManager) { return new StepBuilder("taskletStep", jobRepository) .tasklet((contribution, chunkContext) -> { cleanupService.cleanup(); return RepeatStatus.FINISHED; }, txManager) .build(); }

ItemReader

FlatFileItemReader

@Bean public FlatFileItemReader<Person> csvReader() { return new FlatFileItemReaderBuilder<Person>() .name("personReader") .resource(new ClassPathResource("data/input.csv")) .delimited() .delimiter(",") .names("firstName", "lastName", "email", "age") .linesToSkip(1) // Skip header .fieldSetMapper(new BeanWrapperFieldSetMapper<>() {{ setTargetType(Person.class); }}) .build(); }

JdbcPagingItemReader

@Bean public JdbcPagingItemReader<Person> pagingReader(DataSource dataSource) { Map<String, Order> sortKeys = new HashMap<>(); sortKeys.put("id", Order.ASCENDING);

return new JdbcPagingItemReaderBuilder<Person>()
    .name("pagingReader")
    .dataSource(dataSource)
    .selectClause("SELECT id, first_name, last_name, email")
    .fromClause("FROM persons")
    .whereClause("WHERE status = :status")
    .parameterValues(Map.of("status", "ACTIVE"))
    .sortKeys(sortKeys)
    .rowMapper(new BeanPropertyRowMapper<>(Person.class))
    .pageSize(100)
    .build();

}

JpaPagingItemReader

@Bean public JpaPagingItemReader<Person> jpaReader(EntityManagerFactory emf) { return new JpaPagingItemReaderBuilder<Person>() .name("jpaReader") .entityManagerFactory(emf) .queryString("SELECT p FROM Person p WHERE p.status = :status") .parameterValues(Map.of("status", Status.ACTIVE)) .pageSize(100) .build(); }

ItemProcessor

@Component public class PersonProcessor implements ItemProcessor<Person, Person> {

@Override
public Person process(Person person) throws Exception {
    // Return null to filter out item
    if (!isValid(person)) {
        return null;
    }

    // Transform
    person.setEmail(person.getEmail().toLowerCase());
    person.setFullName(person.getFirstName() + " " + person.getLastName());
    return person;
}

}

// Composite processor @Bean public CompositeItemProcessor<Person, Person> compositeProcessor() { return new CompositeItemProcessorBuilder<Person, Person>() .delegates(List.of( validationProcessor(), transformationProcessor(), enrichmentProcessor() )) .build(); }

ItemWriter

JdbcBatchItemWriter

@Bean public JdbcBatchItemWriter<Person> jdbcWriter(DataSource dataSource) { return new JdbcBatchItemWriterBuilder<Person>() .dataSource(dataSource) .sql("INSERT INTO persons (first_name, last_name, email) VALUES (:firstName, :lastName, :email)") .beanMapped() .build(); }

JpaItemWriter

@Bean public JpaItemWriter<Person> jpaWriter(EntityManagerFactory emf) { JpaItemWriter<Person> writer = new JpaItemWriter<>(); writer.setEntityManagerFactory(emf); writer.setUsePersist(true); // false = merge return writer; }

FlatFileItemWriter

@Bean public FlatFileItemWriter<Person> csvWriter() { return new FlatFileItemWriterBuilder<Person>() .name("personWriter") .resource(new FileSystemResource("output/persons.csv")) .delimited() .delimiter(",") .names("firstName", "lastName", "email") .headerCallback(writer -> writer.write("First Name,Last Name,Email")) .build(); }

Job Parameters

@Bean @StepScope public FlatFileItemReader<Person> parameterizedReader( @Value("#{jobParameters['inputFile']}") String inputFile) { return new FlatFileItemReaderBuilder<Person>() .name("reader") .resource(new FileSystemResource(inputFile)) .delimited() .names("firstName", "lastName", "email") .targetType(Person.class) .build(); }

// Running job with parameters @Service @RequiredArgsConstructor public class JobLauncherService {

private final JobLauncher jobLauncher;
private final Job importJob;

public void runJob(String inputFile, LocalDate date) throws Exception {
    JobParameters params = new JobParametersBuilder()
        .addString("inputFile", inputFile)
        .addLocalDate("date", date)
        .addLong("timestamp", System.currentTimeMillis())
        .toJobParameters();

    JobExecution execution = jobLauncher.run(importJob, params);
    log.info("Job status: {}", execution.getStatus());
}

}

When NOT to Use This Skill

• Real-time processing - Use streaming (Kafka Streams, Flink)

• Simple scheduled tasks - Use spring-scheduling instead

• Microservices data sync - Consider event-driven with messaging

• Small data sets - Batch overhead may not be justified

Anti-Patterns

Anti-Pattern Problem Solution

Large chunk size Memory issues, long transactions Tune chunk size (100-1000)

No skip policy Single error stops job Configure skip for expected errors

Stateful ItemProcessor Thread safety issues Make processor stateless

Ignoring job parameters Can't restart failed jobs Include identifying params

No monitoring Silent failures Configure JobExecutionListener

Single-threaded for large data Slow processing Use partitioning

Quick Troubleshooting

Problem Diagnostic Fix

Job not starting Check job repository Verify database schema

Job fails on restart Check job parameters Add RunIdIncrementer

Chunk processing slow Check commit interval Tune chunk size

Memory issues Monitor heap usage Reduce chunk size, stream data

Skip not working Check skip policy Configure skippable exceptions

Best Practices

• ✅ Use chunk processing for large volumes

• ✅ Configure skip/retry for fault tolerance

• ✅ Use partitioning for parallelism

• ✅ Implement listeners for monitoring

• ✅ Test with JobLauncherTestUtils

• ❌ Don't use chunk size too large

• ❌ Don't ignore errors silently

• ❌ Don't forget job parameters for restart

Reference Documentation

• Spring Batch