asmhatre/spring-boot-demo
1
0
Fork 0
Code Activity

Add spring-batch: jobs, steps, chunk processing and restartability on Boot 4.1

Browse Source 
Companion code for "Spring Batch on Boot 4.1: Jobs, Steps, Chunk Processing and
Restartability". A productImportJob configured three ways by profile against a
poisoned CSV row, run as real java -jar processes (not just JUnit) so the
restart story is genuine: a chunk fails and rolls back, the process exits, a
brand-new JVM against the same file-based H2 database resumes at the exact
next unread row (READ_COUNT 20, not 60) and completes.

Findings the build pins:
- StepBuilder.chunk(int, PlatformTransactionManager) still compiles in Batch
  6.0.5 but returns the legacy SimpleStepBuilder; chunk(int) returns the new
  ChunkOrientedStepBuilder, and only the latter is used here.
- Two different ExecutionContext classes now exist in two different packages
  (infrastructure.item vs core.repository.persistence) with different shapes.
- spring-boot-starter-batch alone gives a resourceless JobRepository that
  forgets every JobInstance the moment the JVM exits; spring-boot-starter-
  batch-jdbc is what makes the restart demo possible at all, demonstrated by
  excluding BatchJdbcAutoConfiguration and watching a "restart" collide with
  the previous run's own data instead of resuming it.
- A migration-guide summary claiming CommandLineJobRunner was removed in 6.0
  is wrong -- javap against the real jar shows @Deprecated(forRemoval=true),
  not removed.
- RepeatStatus moved from core.repeat to infrastructure.repeat, caught by the
  compiler rather than by reading docs.

11 documentation chapters, 10 captured transcripts (unit tests, javap output,
and real two-JVM scenario runs), all regenerated by scripts/run-all.sh.
This commit is contained in:
Claude
2026-09-13 06:24:06 +00:00
parent a9867c0423
commit 6943fc527f
40 changed files with 2001 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
spring-batch/src/main/java/com/ankurm/batch/BatchDemoApplication.java
+11
View File
@@ -0,0 +1,11 @@
package com.ankurm.batch;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class BatchDemoApplication {
public static void main(String[] args) {
SpringApplication.run(BatchDemoApplication.class, args);
}
}
spring-batch/src/main/java/com/ankurm/batch/config/BatchConfig.java
+189
View File
@@ -0,0 +1,189 @@
package com.ankurm.batch.config;
import com.ankurm.batch.domain.Product;
import com.ankurm.batch.processing.ProductValidatingProcessor;
import org.springframework.batch.core.job.Job;
import org.springframework.batch.core.job.builder.JobBuilder;
import org.springframework.batch.core.listener.SkipListener;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.Step;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.infrastructure.item.ItemProcessor;
import org.springframework.batch.infrastructure.item.ItemWriter;
import org.springframework.batch.infrastructure.item.database.builder.JdbcBatchItemWriterBuilder;
import org.springframework.batch.infrastructure.item.file.builder.FlatFileItemReaderBuilder;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Profile;
import org.springframework.core.io.Resource;
import org.springframework.dao.DataIntegrityViolationException;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.transaction.PlatformTransactionManager;
/**
* Wires the same "import products, then report a count" job three different ways, selected by
* Spring profile:
*
* <ul>
* <li>{@code clean} &mdash; the happy path, chapters 1-3</li>
* <li>{@code broken} &mdash; no fault tolerance, chapter 7 (restartability)</li>
* <li>{@code skip} &mdash; {@code faultTolerant().skip(...)}, chapter 8 (skip vs. restart)</li>
* </ul>
*
* All three read {@link #productReader}, run it through {@link ProductValidatingProcessor}, and
* write with {@link #productWriter}; only the step's fault-tolerance configuration and the input
* file differ. See {@code docs/02-anatomy-of-a-job.md} for why the job and step beans look like
* this rather than the {@code JobBuilderFactory} / {@code StepBuilderFactory} shape older
* tutorials still show (that pair was removed years before this article; it is not a Boot 4.1
* surprise, just a dead end still copy-pasted in 2026).
*/
@Configuration
public class BatchConfig {
@Value("${import.file:classpath:data/products.csv}")
private Resource inputFile;
// ---- shared reader / processor / writer -----------------------------------------------
@Bean
public org.springframework.batch.infrastructure.item.file.FlatFileItemReader<Product> productReader() {
return new FlatFileItemReaderBuilder<Product>()
.name("productReader")
.resource(inputFile)
.linesToSkip(1)
.delimited().delimiter(",").names("sku", "name", "priceCents")
.fieldSetMapper(fs -> new Product(fs.readString("sku"), fs.readString("name"), fs.readLong("priceCents")))
.build();
}
@Bean
public ItemProcessor<Product, Product> productProcessor() {
return new ProductValidatingProcessor();
}
/**
* Deliberately NOT {@code beanMapped()}: {@code BeanPropertySqlParameterSource} looks for
* {@code getSku()}, {@code getName()}, {@code getPriceCents()} via standard JavaBean
* introspection, and a Java record's accessors are {@code sku()}, {@code name()},
* {@code priceCents()} &mdash; no {@code get} prefix. {@code docs/06-jdbc-writer-and-records.md}
* has the transcript of {@code beanMapped()} silently writing every column as NULL against
* this exact record before this was caught.
*/
@Bean
public ItemWriter<Product> productWriter(JdbcTemplate jdbcTemplate) {
return new JdbcBatchItemWriterBuilder<Product>()
.dataSource(jdbcTemplate.getDataSource())
.sql("INSERT INTO PRODUCT (sku, name, price_cents) VALUES (?, ?, ?)")
.itemPreparedStatementSetter((item, ps) -> {
ps.setString(1, item.sku());
ps.setString(2, item.name());
ps.setLong(3, item.priceCents());
})
.assertUpdates(true)
.build();
}
@Bean
public Tasklet reportTasklet(JdbcTemplate jdbcTemplate) {
return (contribution, chunkContext) -> {
int count = jdbcTemplate.queryForObject("SELECT COUNT(*) FROM PRODUCT", Integer.class);
System.out.println("REPORT: " + count + " products now in the PRODUCT table");
return org.springframework.batch.infrastructure.repeat.RepeatStatus.FINISHED;
};
}
// ---- clean profile: happy path, no fault tolerance needed ------------------------------
@Bean
@Profile("clean")
public Step importStepClean(JobRepository jobRepository, PlatformTransactionManager transactionManager,
org.springframework.batch.infrastructure.item.file.FlatFileItemReader<Product> productReader,
ItemProcessor<Product, Product> productProcessor, ItemWriter<Product> productWriter) {
return new StepBuilder("importStep", jobRepository)
.<Product, Product>chunk(10)
.transactionManager(transactionManager)
.reader(productReader)
.processor(productProcessor)
.writer(productWriter)
.build();
}
// ---- broken profile: no fault tolerance -> the whole chunk fails, job FAILS ------------
@Bean
@Profile("broken")
public Step importStepBroken(JobRepository jobRepository, PlatformTransactionManager transactionManager,
org.springframework.batch.infrastructure.item.file.FlatFileItemReader<Product> productReader,
ItemProcessor<Product, Product> productProcessor, ItemWriter<Product> productWriter) {
return new StepBuilder("importStep", jobRepository)
.<Product, Product>chunk(10)
.transactionManager(transactionManager)
.reader(productReader)
.processor(productProcessor)
.writer(productWriter)
.build();
}
// ---- skip profile: faultTolerant + skip -> the bad item is skipped, job COMPLETES -----
@Bean
@Profile("skip")
public Step importStepSkip(JobRepository jobRepository, PlatformTransactionManager transactionManager,
org.springframework.batch.infrastructure.item.file.FlatFileItemReader<Product> productReader,
ItemProcessor<Product, Product> productProcessor, ItemWriter<Product> productWriter) {
return new StepBuilder("importStep", jobRepository)
.<Product, Product>chunk(10)
.transactionManager(transactionManager)
.reader(productReader)
.processor(productProcessor)
.writer(productWriter)
.faultTolerant()
.skip(DataIntegrityViolationException.class)
.skipLimit(3)
.listener(skipListener())
.build();
}
@Bean
public SkipListener<Product, Product> skipListener() {
return new SkipListener<>() {
@Override
public void onSkipInWrite(Product item, Throwable t) {
System.out.println("SKIPPED on write: " + item.sku() + " (" + t.getClass().getSimpleName() + ": " + t.getMessage() + ")");
}
};
}
// ---- reportStep: always runs, even if the step before it was already COMPLETED --------
@Bean
public Step reportStep(JobRepository jobRepository, PlatformTransactionManager transactionManager,
Tasklet reportTasklet) {
return new StepBuilder("reportStep", jobRepository)
.tasklet(reportTasklet, transactionManager)
.allowStartIfComplete(true)
.build();
}
// ---- one Job bean per profile, all built the same way ----------------------------------
@Bean
@Profile("clean")
public Job productImportJobClean(JobRepository jobRepository, Step importStepClean, Step reportStep) {
return new JobBuilder("productImportJob", jobRepository).start(importStepClean).next(reportStep).build();
}
@Bean
@Profile("broken")
public Job productImportJobBroken(JobRepository jobRepository, Step importStepBroken, Step reportStep) {
return new JobBuilder("productImportJob", jobRepository).start(importStepBroken).next(reportStep).build();
}
@Bean
@Profile("skip")
public Job productImportJobSkip(JobRepository jobRepository, Step importStepSkip, Step reportStep) {
return new JobBuilder("productImportJob", jobRepository).start(importStepSkip).next(reportStep).build();
}
}
spring-batch/src/main/java/com/ankurm/batch/domain/Product.java
+12
View File
@@ -0,0 +1,12 @@
package com.ankurm.batch.domain;
/**
* One validated row, ready to be written to the {@code PRODUCT} table.
*
* <p>See {@code docs/03-chunk-oriented-processing.md} for how instances of this record move
* through the reader &rarr; processor &rarr; writer pipeline in chunks, and
* {@code docs/06-jdbc-writer-and-records.md} for why the writer below does not use
* {@code beanMapped()} against this record.
*/
public record Product(String sku, String name, long priceCents) {
}
spring-batch/src/main/java/com/ankurm/batch/processing/ProductValidatingProcessor.java
+35
View File
@@ -0,0 +1,35 @@
package com.ankurm.batch.processing;
import com.ankurm.batch.domain.Product;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.infrastructure.item.ItemProcessor;
import java.util.regex.Pattern;
/**
* Chapter 4 ({@code docs/04-item-processor-as-filter.md}): a processor that returns
* {@code null} for a row it does not like. Spring Batch treats a {@code null} return as
* "filter this item" &mdash; it is counted separately from both reads and writes, and it
* never reaches the writer or the database. This is deliberately NOT how the poisoned
* duplicate SKU is handled: that one is format-valid and only fails at the database, which is
* the point of chapter 7.
*/
public class ProductValidatingProcessor implements ItemProcessor<Product, Product> {
private static final Log log = LogFactory.getLog(ProductValidatingProcessor.class);
private static final Pattern SKU_PATTERN = Pattern.compile("^[A-Z]{3}-\\d{4}$");
@Override
public Product process(Product item) {
if (!SKU_PATTERN.matcher(item.sku()).matches()) {
log.warn("filtering " + item.sku() + ": does not match " + SKU_PATTERN.pattern());
return null;
}
if (item.priceCents() <= 0) {
log.warn("filtering " + item.sku() + ": priceCents must be positive, was " + item.priceCents());
return null;
}
return item;
}
}
spring-batch/src/main/java/com/ankurm/batch/runner/ImportRunner.java
+50
View File
@@ -0,0 +1,50 @@
package com.ankurm.batch.runner;
import org.springframework.batch.core.job.Job;
import org.springframework.batch.core.job.parameters.JobParametersBuilder;
import org.springframework.batch.core.launch.JobOperator;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;
/**
* Chapter 5 ({@code docs/05-launching-and-jobparameters.md}) and chapter 7
* ({@code docs/07-restartability.md}): calling {@link JobOperator#start} with the SAME
* identifying job parameter every time is, on its own, the restart mechanism. There is no
* separate "restart" button here: if a JobInstance with these identifying parameters already
* exists and its last execution did not complete, {@code start} runs a new JobExecution against
* that same instance and Spring Batch resumes each step from where its own ExecutionContext says
* it left off. If the last execution DID complete, {@code start} throws
* {@link org.springframework.batch.core.launch.JobInstanceAlreadyCompleteException}, which this
* runner treats as "nothing to do" rather than a failure.
*
* <p>{@code spring.batch.job.enabled} is left at its Boot default of {@code true}, but the
* auto-configured {@code JobLauncherApplicationRunner} would launch every Job bean in the
* context using default parameters, which collides with the explicit control this module wants
* to demonstrate -- see {@code application.yml}, where it is turned off.
*/
@Component
public class ImportRunner implements ApplicationRunner {
private final JobOperator jobOperator;
private final Job job;
public ImportRunner(JobOperator jobOperator, Job job) {
this.jobOperator = jobOperator;
this.job = job;
}
@Override
public void run(ApplicationArguments args) throws Exception {
var params = new JobParametersBuilder()
.addString("batch.run", "demo") // identifying: same value = same JobInstance = restart target
.toJobParameters();
try {
var execution = jobOperator.start(job, params);
System.out.println("JOB FINISHED: status=" + execution.getStatus()
+ " exitCode=" + execution.getExitStatus().getExitCode());
} catch (org.springframework.batch.core.launch.JobInstanceAlreadyCompleteException e) {
System.out.println("JOB ALREADY COMPLETE: " + e.getMessage());
}
}
}