Add spring-batch: jobs, steps, chunk processing and restartability on Boot 4.1
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. Fixed after push: three dead docs.spring.io links in the doc chapters (readersAndWriters/* and chunk-oriented-processing/*.html paths moved when Spring Batch 6 reorganized its reference docs; corrected to the current readers-and-writers/*, processor.html and chunk-oriented-processing.html paths, verified 200 via curl before committing). Co-Authored-By: Claude Sonnet 5 <[email protected]> Claude-Session: https://claude.ai/code/session_019DXsJ1zpikbA1MQJN6RqFA
This commit is contained in:
Claude
@@ -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);
|
||||
}
|
||||
}
|
||||
@@ -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} — the happy path, chapters 1-3</li>
|
||||
* <li>{@code broken} — no fault tolerance, chapter 7 (restartability)</li>
|
||||
* <li>{@code skip} — {@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()} — 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();
|
||||
}
|
||||
}
|
||||
@@ -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 → processor → 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) {
|
||||
}
|
||||
+35
@@ -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" — 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;
|
||||
}
|
||||
}
|
||||
@@ -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());
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,22 @@
|
||||
spring:
|
||||
batch:
|
||||
job:
|
||||
# ImportRunner drives the job explicitly through JobOperator. The auto-configured
|
||||
# JobLauncherApplicationRunner would ALSO launch every Job bean it finds, using
|
||||
# parameterless defaults -- turning this off avoids running the job twice on every startup.
|
||||
enabled: false
|
||||
jdbc:
|
||||
initialize-schema: always
|
||||
datasource:
|
||||
url: jdbc:h2:file:./data/batchdb;AUTO_SERVER=TRUE
|
||||
username: sa
|
||||
password: ""
|
||||
driver-class-name: org.h2.Driver
|
||||
sql:
|
||||
init:
|
||||
mode: always
|
||||
schema-locations: classpath:schema.sql
|
||||
|
||||
logging:
|
||||
level:
|
||||
org.springframework.batch: INFO
|
||||
@@ -0,0 +1,61 @@
|
||||
sku,name,priceCents
|
||||
ABC-0001,Widget 1,1001
|
||||
ABC-0002,Widget 2,1002
|
||||
ABC-0003,Widget 3,1003
|
||||
ABC-0004,Widget 4,1004
|
||||
ABC-0005,Widget 5,1005
|
||||
ABC-0006,Widget 6,1006
|
||||
ABC-0007,Widget 7,1007
|
||||
ABC-0008,Widget 8,1008
|
||||
ABC-0009,Widget 9,1009
|
||||
ABC-0010,Widget 10,1010
|
||||
ABC-0011,Widget 11,1011
|
||||
abc-0012,Widget 12,1012
|
||||
ABC-0013,Widget 13,1013
|
||||
ABC-0014,Widget 14,1014
|
||||
ABC-0015,Widget 15,1015
|
||||
ABC-0016,Widget 16,1016
|
||||
ABC-0017,Widget 17,1017
|
||||
ABC-0018,Widget 18,1018
|
||||
ABC-0019,Widget 19,1019
|
||||
ABC-0020,Widget 20,1020
|
||||
ABC-0021,Widget 21,1021
|
||||
ABC-0022,Widget 22,1022
|
||||
ABC-0023,Widget 23,1023
|
||||
ABC-0024,Widget 24,1024
|
||||
ABC-0025,Widget 25,1025
|
||||
ABC-0026,Widget 26,1026
|
||||
ABC-0027,Widget 27,1027
|
||||
ABC-0028,Widget 28,1028
|
||||
ABC-0029,Widget 29,1029
|
||||
ABC-0030,Widget 30,1030
|
||||
ABC-0031,Widget 31,1031
|
||||
ABC-0032,Widget 32,1032
|
||||
ABC-0033,Widget 33,0
|
||||
ABC-0034,Widget 34,1034
|
||||
ABC-0035,Widget 35,1035
|
||||
ABC-0036,Widget 36,1036
|
||||
ABC-0037,Widget 37,1037
|
||||
ABC-0038,Widget 38,1038
|
||||
ABC-0039,Widget 39,1039
|
||||
ABC-0040,Widget 40,1040
|
||||
ABC-0041,Widget 41,1041
|
||||
ABC-0042,Widget 42,1042
|
||||
ABC-0043,Widget 43,1043
|
||||
ABC-0044,Widget 44,1044
|
||||
ABC-0045,Widget 45,1045
|
||||
ABC-0046,Widget 46,1046
|
||||
ABC-0005,Widget 47,1047
|
||||
ABC-0048,Widget 48,1048
|
||||
ABC-0049,Widget 49,1049
|
||||
ABC-0050,Widget 50,1050
|
||||
ABC-0051,Widget 51,1051
|
||||
ABC-0052,Widget 52,1052
|
||||
ABC-0053,Widget 53,1053
|
||||
ABC-0054,Widget 54,1054
|
||||
ABC-0055,Widget 55,1055
|
||||
ABC-0056,Widget 56,1056
|
||||
ABC-0057,Widget 57,1057
|
||||
ABC-0058,Widget 58,1058
|
||||
ABC-0059,Widget 59,1059
|
||||
ABC-0060,Widget 60,1060
|
||||
|
@@ -0,0 +1,61 @@
|
||||
sku,name,priceCents
|
||||
ABC-0001,Widget 1,1001
|
||||
ABC-0002,Widget 2,1002
|
||||
ABC-0003,Widget 3,1003
|
||||
ABC-0004,Widget 4,1004
|
||||
ABC-0005,Widget 5,1005
|
||||
ABC-0006,Widget 6,1006
|
||||
ABC-0007,Widget 7,1007
|
||||
ABC-0008,Widget 8,1008
|
||||
ABC-0009,Widget 9,1009
|
||||
ABC-0010,Widget 10,1010
|
||||
ABC-0011,Widget 11,1011
|
||||
abc-0012,Widget 12,1012
|
||||
ABC-0013,Widget 13,1013
|
||||
ABC-0014,Widget 14,1014
|
||||
ABC-0015,Widget 15,1015
|
||||
ABC-0016,Widget 16,1016
|
||||
ABC-0017,Widget 17,1017
|
||||
ABC-0018,Widget 18,1018
|
||||
ABC-0019,Widget 19,1019
|
||||
ABC-0020,Widget 20,1020
|
||||
ABC-0021,Widget 21,1021
|
||||
ABC-0022,Widget 22,1022
|
||||
ABC-0023,Widget 23,1023
|
||||
ABC-0024,Widget 24,1024
|
||||
ABC-0025,Widget 25,1025
|
||||
ABC-0026,Widget 26,1026
|
||||
ABC-0027,Widget 27,1027
|
||||
ABC-0028,Widget 28,1028
|
||||
ABC-0029,Widget 29,1029
|
||||
ABC-0030,Widget 30,1030
|
||||
ABC-0031,Widget 31,1031
|
||||
ABC-0032,Widget 32,1032
|
||||
ABC-0033,Widget 33,0
|
||||
ABC-0034,Widget 34,1034
|
||||
ABC-0035,Widget 35,1035
|
||||
ABC-0036,Widget 36,1036
|
||||
ABC-0037,Widget 37,1037
|
||||
ABC-0038,Widget 38,1038
|
||||
ABC-0039,Widget 39,1039
|
||||
ABC-0040,Widget 40,1040
|
||||
ABC-0041,Widget 41,1041
|
||||
ABC-0042,Widget 42,1042
|
||||
ABC-0043,Widget 43,1043
|
||||
ABC-0044,Widget 44,1044
|
||||
ABC-0045,Widget 45,1045
|
||||
ABC-0046,Widget 46,1046
|
||||
ABC-0047,Widget 47,1047
|
||||
ABC-0048,Widget 48,1048
|
||||
ABC-0049,Widget 49,1049
|
||||
ABC-0050,Widget 50,1050
|
||||
ABC-0051,Widget 51,1051
|
||||
ABC-0052,Widget 52,1052
|
||||
ABC-0053,Widget 53,1053
|
||||
ABC-0054,Widget 54,1054
|
||||
ABC-0055,Widget 55,1055
|
||||
ABC-0056,Widget 56,1056
|
||||
ABC-0057,Widget 57,1057
|
||||
ABC-0058,Widget 58,1058
|
||||
ABC-0059,Widget 59,1059
|
||||
ABC-0060,Widget 60,1060
|
||||
|
@@ -0,0 +1,10 @@
|
||||
-- Application table, separate from the BATCH_* tables that BatchJdbcAutoConfiguration
|
||||
-- creates via spring.batch.jdbc.initialize-schema. The UNIQUE constraint on sku is what turns
|
||||
-- the poisoned duplicate row into a real DataIntegrityViolationException at chunk-commit time
|
||||
-- rather than something the processor could have caught by looking at one row in isolation.
|
||||
CREATE TABLE IF NOT EXISTS PRODUCT (
|
||||
id BIGINT AUTO_INCREMENT PRIMARY KEY,
|
||||
sku VARCHAR(16) NOT NULL UNIQUE,
|
||||
name VARCHAR(200) NOT NULL,
|
||||
price_cents BIGINT NOT NULL
|
||||
);
|
||||
@@ -0,0 +1,63 @@
|
||||
package com.ankurm.batch;
|
||||
|
||||
import com.ankurm.batch.domain.Product;
|
||||
import com.ankurm.batch.processing.ProductValidatingProcessor;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.springframework.dao.DataIntegrityViolationException;
|
||||
import org.springframework.dao.DuplicateKeyException;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
/**
|
||||
* Chapter 4 ({@code docs/04-item-processor-as-filter.md}): a null return from
|
||||
* {@link org.springframework.batch.infrastructure.item.ItemProcessor#process} is a filter, not
|
||||
* a skip. These assertions pin exactly which rows the pipeline's processor removes and which it
|
||||
* lets through, independent of Spring Batch or the database.
|
||||
*/
|
||||
class ProductValidatingProcessorTest {
|
||||
|
||||
private final ProductValidatingProcessor processor = new ProductValidatingProcessor();
|
||||
|
||||
@Test
|
||||
void passesAWellFormedRow() {
|
||||
try (Transcript t = new Transcript("01-processor-filter.txt",
|
||||
"The processor as a filter: null means skip, not fail")) {
|
||||
|
||||
Product ok = new Product("ABC-0001", "Widget 1", 1001);
|
||||
Product result = processor.process(ok);
|
||||
t.line("input : %s", ok);
|
||||
t.line("result : %s", result);
|
||||
assertThat(result).isEqualTo(ok);
|
||||
|
||||
t.blank();
|
||||
Product badSku = new Product("abc-0012", "Widget 12", 1012);
|
||||
t.line("input : %s <- lowercase sku, fails ^[A-Z]{3}-\\d{4}$", badSku);
|
||||
t.line("result : %s", processor.process(badSku));
|
||||
assertThat(processor.process(badSku)).isNull();
|
||||
|
||||
t.blank();
|
||||
Product zeroPrice = new Product("ABC-0033", "Widget 33", 0);
|
||||
t.line("input : %s <- priceCents is zero", zeroPrice);
|
||||
t.line("result : %s", processor.process(zeroPrice));
|
||||
assertThat(processor.process(zeroPrice)).isNull();
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void duplicateKeyExceptionIsADataIntegrityViolationException() {
|
||||
try (Transcript t = new Transcript("02-exception-hierarchy.txt",
|
||||
"Why skip(DataIntegrityViolationException.class) also catches the duplicate-key case")) {
|
||||
|
||||
DuplicateKeyException dup = new DuplicateKeyException("Unique index violation");
|
||||
t.line("thrown type : %s", dup.getClass().getName());
|
||||
t.line("is a DataIntegrityViolationException? %b", dup instanceof DataIntegrityViolationException);
|
||||
t.line("");
|
||||
t.line("H2's JdbcBatchUpdateException on a UNIQUE-constraint violation is translated by");
|
||||
t.line("Spring's SQLExceptionSubclassTranslator into DuplicateKeyException, which extends");
|
||||
t.line("DataIntegrityViolationException. A skip policy configured against the parent class");
|
||||
t.line("catches the subclass too -- see docs/07-restartability.md and docs/08-skip-vs-restart.md.");
|
||||
|
||||
assertThat(dup).isInstanceOf(DataIntegrityViolationException.class);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,52 @@
|
||||
package com.ankurm.batch;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.io.PrintWriter;
|
||||
import java.io.StringWriter;
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Path;
|
||||
|
||||
/**
|
||||
* Writes a numbered transcript under {@code docs/output/} and echoes it to the console.
|
||||
* Every console block quoted in the article comes out of one of these files verbatim.
|
||||
*/
|
||||
public final class Transcript implements AutoCloseable {
|
||||
|
||||
private final Path path;
|
||||
private final StringWriter buffer = new StringWriter();
|
||||
private final PrintWriter out = new PrintWriter(buffer);
|
||||
|
||||
public Transcript(String fileName, String title) {
|
||||
this.path = Path.of("docs", "output", fileName);
|
||||
out.println("# " + title);
|
||||
out.println();
|
||||
}
|
||||
|
||||
public Transcript line(String format, Object... args) {
|
||||
out.println(args.length == 0 ? format : String.format(format, args));
|
||||
return this;
|
||||
}
|
||||
|
||||
public Transcript blank() {
|
||||
out.println();
|
||||
return this;
|
||||
}
|
||||
|
||||
public Transcript section(String heading) {
|
||||
out.println();
|
||||
out.println("--- " + heading + " ---");
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
out.flush();
|
||||
try {
|
||||
Files.createDirectories(path.getParent());
|
||||
Files.writeString(path, buffer.toString());
|
||||
} catch (IOException e) {
|
||||
throw new IllegalStateException("could not write " + path, e);
|
||||
}
|
||||
System.out.print(buffer);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user