diff --git a/.gitignore b/.gitignore index bdeb402..9262c83 100644 --- a/.gitignore +++ b/.gitignore @@ -13,3 +13,6 @@ spring-boot-startup-time/src/main/java/com/ankurm/startup/bulk/ # spring-batch: file-based H2 databases created by running the demo spring-batch/data/ spring-batch/scenario-data/ + +# spring-batch-partitioning: generated shard CSVs and file-based H2 databases +spring-batch-partitioning/data/ diff --git a/README.md b/README.md index 6dda18b..119b28a 100644 --- a/README.md +++ b/README.md @@ -20,6 +20,7 @@ files. | [`resilience/`](resilience) | [Spring Framework 7's Built-in Resilience: @Retryable, @ConcurrencyLimit, and What's Left for Resilience4j](https://ankurm.com/spring-framework-7-retryable-concurrencylimit-resilience4j/) | `@Retryable` and `@ConcurrencyLimit` counted invocation by invocation, retries inside transactions, where Resilience4j still earns its place, migrating from Spring Retry | | [`caching/`](caching) | [The Spring Cache Abstraction: @Cacheable, @CacheEvict, Key Generators and the Self-Invocation Trap](https://ankurm.com/spring-cache-abstraction-cacheable-cacheevict-self-invocation-trap/) | the self-invocation trap measured four ways, the key collision `SimpleKeyGenerator` makes easy, eviction timing under a thrown exception, a rollback the cache keeps, and where this sits next to Hibernate's L2 cache | | [`spring-batch/`](spring-batch) | [Spring Batch on Boot 4.1: Jobs, Steps, Chunk Processing and Restartability](https://ankurm.com/) | a job that fails mid-chunk and resumes exactly where it left off across two separate JVMs, skip vs. restart on the same poisoned row, the resourceless job repository that forgets a restart ever happened, and the `chunk(int)` vs `chunk(int, tx)` builder split | +| [`spring-batch-partitioning/`](spring-batch-partitioning) | [Spring Batch Partitioning and Parallel Steps: Scaling a 10-Million-Row Job](https://ankurm.com/) | the real grid-size sweep at 10M and 300K rows (best speedup 1.42x, on 2 cores), `MultiResourcePartitioner` ignoring gridSize entirely, a rejected partition's `StepExecution` stuck at `STARTING` forever, and Spring Batch 6.0's new `JobOperator#recover` unsticking it | Articles whose text is kept here rather than only on the blog have it under `/post/` — `post.md` for the body and `meta.md` for the title, excerpt and diff --git a/spring-batch-partitioning/README.md b/spring-batch-partitioning/README.md new file mode 100644 index 0000000..f3ea07e --- /dev/null +++ b/spring-batch-partitioning/README.md @@ -0,0 +1,73 @@ +# spring-batch-partitioning + +Companion code for **[Spring Batch Partitioning and Parallel Steps: Scaling a 10-Million-Row Job](https://ankurm.com/)** +on [ankurm.com](https://ankurm.com). + +Verified against Spring Boot **4.1.1**, Spring Batch **6.0.5**, Spring Framework **7.0.9**, on +Temurin JDK **25.0.4.1+1**, on a 2-vCPU sandbox. + +One job, `orderRiskJob`: partition a directory of pre-sharded order CSVs across worker threads, +score every order for risk with a deliberately CPU-bound processor, write the result to +`ORDER_RISK_SUMMARY`. There is no separately-coded single-threaded baseline — `--partition.grid-size=1` +against a one-shard directory runs the identical code path as any other grid size (see +[chapter 2](docs/02-anatomy-of-a-partitioned-step.md)), so every number below differs by exactly +one variable. + +| Run | What it demonstrates | Docs | +|---|---|---| +| `--partition.grid-size=1` against a 1-shard dir | Single-threaded baseline, same code path | [ch. 1](docs/01-the-problem-and-mental-model.md), [ch. 2](docs/02-anatomy-of-a-partitioned-step.md) | +| `--partition.grid-size=2/4/8` against matching shard dirs | Partitioned scaling, and where it stops helping | [ch. 3](docs/03-what-gridsize-actually-controls.md), [ch. 10](docs/10-scaling-sensitivity-to-data-size.md) | +| `--spring.profiles.active=reject` | Undersized pool + `AbortPolicy`: partitions rejected, `StepExecution`s stuck at `STARTING` forever | [ch. 7](docs/07-the-rejectedexecutionexception.md) | +| `--spring.profiles.active=recover --recover.job-execution-id=N` | Spring Batch 6.0's `JobOperator#recover`, then a normal restart that reruns only the failed partitions | [ch. 8](docs/08-restart-reruns-only-the-failed-partition.md), [ch. 9](docs/09-jobexecutionalreadyrunning-and-recover.md) | + +## Documentation chapters + +1. [The problem, and the smallest correct mental model](docs/01-the-problem-and-mental-model.md) +2. [The anatomy of a partitioned step](docs/02-anatomy-of-a-partitioned-step.md) +3. [What gridSize actually controls](docs/03-what-gridsize-actually-controls.md) +4. [The writer, the beanMapped trap, and finding the partition's own name](docs/04-the-writer-and-the-beanmapped-trap.md) +5. [The diagnostic endpoint](docs/05-the-diagnostic-endpoint.md) +6. [Why this module's work is CPU-bound, not I/O-bound](docs/06-why-cpu-bound-not-io-bound.md) +7. [The failure that does not look like a failure: rejected partitions](docs/07-the-rejectedexecutionexception.md) +8. [Restart reruns only the failed partition — proved, not assumed](docs/08-restart-reruns-only-the-failed-partition.md) +9. [JobExecutionAlreadyRunningException, forever — and recover()](docs/09-jobexecutionalreadyrunning-and-recover.md) +10. [Scaling sensitivity to data size, and the honest ceiling](docs/10-scaling-sensitivity-to-data-size.md) +11. [Production checklist](docs/11-production-checklist.md) + +## Captured output + +Everything under [`docs/output/`](docs/output) was produced by a real run (or a real `mvn test`) +and is quoted verbatim in the article and the chapters above: + +| File | What produced it | +|---|---| +| `01-processor-determinism.txt`, `02-gridsize-ignored.txt` | JUnit tests, via `mvn test` | +| `03-package-repackaging-javap.txt` | `javap` / `unzip -l` against the real 6.0.5 and 5.2.6 jars | +| `04-enforceuniquemethods-error.txt` | A real startup failure, first draft of `BatchConfig` | +| `05-happy-path-4-partitions.txt` | 4 shards, gridSize 4, plus the diagnostic endpoint | +| `06-rejected-partitions-stuck.txt`, `07-restart-throws-alreadyrunning.txt`, `08-recover-then-restart.txt` | The `reject` profile, a failed restart attempt, then the `recover` profile, all against the same H2 file across separate JVMs | +| `09-full-scale-throughput.txt` | The full grid-size sweep at 10,000,000 rows and at 300,000 rows | + +## Running it + +Needs a JDK 25 and Maven 3.9, plus Python 3 for the data generator. + +```bash +export JAVA_HOME=/path/to/jdk-25 +mvn -DskipTests package +python3 scripts/generate-shards.py ./data/shards 10000000 4 # 4 shard files, 2.5M rows each +java -jar target/spring-batch-partitioning-1.0.0.jar --partition.shards-dir=./data/shards --partition.grid-size=4 +``` + +`GET http://localhost:8081/batch/partitions/{jobExecutionId}` (or `/batch/partitions/latest`) +shows which thread ran which partition, and for how long — see +[chapter 5](docs/05-the-diagnostic-endpoint.md). + +`scripts/generate-shards.py [--corrupt-shard N] [--seed S]` produces the +sharded CSVs any of the above commands read; the same seed produces byte-identical row content +regardless of how many shards it is split into, which is what makes the grid-size comparisons in +[chapter 10](docs/10-scaling-sensitivity-to-data-size.md) apples-to-apples. + +## Licence + +MIT — see the repository [LICENSE](../LICENSE). diff --git a/spring-batch-partitioning/docs/01-the-problem-and-mental-model.md b/spring-batch-partitioning/docs/01-the-problem-and-mental-model.md new file mode 100644 index 0000000..39126e5 --- /dev/null +++ b/spring-batch-partitioning/docs/01-the-problem-and-mental-model.md @@ -0,0 +1,82 @@ +# 1. The problem, and the smallest correct mental model + +[README](../README.md) | [Next: Anatomy of a partitioned step →](02-anatomy-of-a-partitioned-step.md) + +## The problem + +A chunk-oriented Spring Batch step is already fast: it reads an item, processes it, and only +touches the database once per chunk, not once per row. But it is still **one reader, one +processor, one writer, one thread.** Give it 10 million rows of CPU-bound scoring work and it +will get through all of them correctly — restartably, with fault tolerance, with every +guarantee chunk-oriented processing gives you — using exactly one CPU core, for as long as +that takes. On the two-core sandbox this module was built and measured on, that is about 70 +seconds for 10 million rows (see +[`docs/output/09-full-scale-throughput.txt`](output/09-full-scale-throughput.txt)). On a job with +real per-row work, or ten times the rows, "one thread" stops being a detail and starts being the +bottleneck. + +Partitioning is Spring Batch's answer to "make more than one thread do this work, without +throwing away restartability." It does not change chunk-oriented processing at all — it +takes the *step itself* and hands several copies of it, each with a different slice of input, to +separate threads (or, wired differently, to separate JVMs on separate machines). Each copy is a +completely ordinary chunk-oriented step: its own reader, its own processor, its own writer, its +own transaction, its own restart bookkeeping in the job repository. + +## The smallest correct mental model + +There are two steps in a partitioned job, not one, and they are structurally different: + +- **The manager step** does no item processing. Its entire job is to ask a `Partitioner` for a + set of input descriptions (in this module: which shard CSV file), hand each one to a worker + thread as a separate `StepExecution`, wait for all of them, and roll the results up into one + outcome. +- **The worker step** is the chunk-oriented step you already know from + [the earlier `spring-batch` article](https://ankurm.com/) — reader, processor, writer, + fault tolerance, all of it — run once per partition, against only that partition's slice + of the input. + +
+ + + +ordersManagerStep +Partitioner.partition(gridSize) returns one ExecutionContext per shard file +fileName=shard-00.csv +fileName=shard-01.csv +fileName=shard-02.csv +fileName=shard-03.csv +worker steppartition0, thread A +worker steppartition1, thread B +worker steppartition2, thread C +worker steppartition3, thread D + +ORDER_RISK_SUMMARY (shared H2 file) + +
+ +The `Partitioner` never sees a row of data. It only produces *descriptions* of work — in +this module, one file path per partition, via the built-in `MultiResourcePartitioner`. The actual +CSV parsing, risk scoring, and database writing all happen inside the worker step, four separate +times, on four separate threads, each against its own file. See +[chapter 3](03-what-gridsize-actually-controls.md) for exactly how many partitions actually run, +which is a more interesting question than it sounds. + +## What this buys you, and what it does not + +Partitioning parallelizes CPU-bound and I/O-bound work across threads (or machines) while keeping +every restart, skip, and retry guarantee chunk-oriented processing already gives a single step +— each partition restarts independently, as [chapter 8](08-restart-reruns-only-the-failed-partition.md) +demonstrates concretely. It does **not** automatically parallelize a shared bottleneck: four +threads writing to the same single-writer embedded database do not write four times as fast just +because four threads are asking. [Chapter 6](06-why-cpu-bound-not-io-bound.md) and +[chapter 10](10-scaling-sensitivity-to-data-size.md) measure exactly how much that shortfall is, +on this hardware, with this writer. + +## Going deeper + +- Manager/worker terminology and the full SPI: + [Spring Batch reference — Scaling and Parallel Processing](https://docs.spring.io/spring-batch/reference/scalability.html) (`rel="nofollow"`). +- The chunk-oriented step this builds on: + [Spring Batch on Boot 4.1: Jobs, Steps, Chunk Processing and Restartability](https://ankurm.com/). + +[Next: Anatomy of a partitioned step →](02-anatomy-of-a-partitioned-step.md) diff --git a/spring-batch-partitioning/docs/02-anatomy-of-a-partitioned-step.md b/spring-batch-partitioning/docs/02-anatomy-of-a-partitioned-step.md new file mode 100644 index 0000000..2a56a97 --- /dev/null +++ b/spring-batch-partitioning/docs/02-anatomy-of-a-partitioned-step.md @@ -0,0 +1,105 @@ +# 2. The anatomy of a partitioned step + +[← Previous](01-the-problem-and-mental-model.md) | [README](../README.md) | [Next: What gridSize actually controls →](03-what-gridsize-actually-controls.md) + +The whole manager/worker wiring is four beans in +[`BatchConfig`](../src/main/java/com/ankurm/batchpartition/config/BatchConfig.java): + +```java +@Bean +public Partitioner partitioner() throws IOException { + var resolver = new PathMatchingResourcePatternResolver(); + Resource[] resources = resolver.getResources("file:" + shardsDir + "/*.csv"); + // ... sort, then: + var partitioner = new MultiResourcePartitioner(); + partitioner.setResources(resources); + partitioner.setKeyName("fileName"); + return partitioner; +} + +@Bean +public Step managerStep(JobRepository jobRepository, Partitioner partitioner, Step workerStep, + TaskExecutor partitionTaskExecutor) { + return new StepBuilder("ordersManagerStep", jobRepository) + .partitioner("ordersWorkerStep", partitioner) + .step(workerStep) + .gridSize(gridSize) + .taskExecutor(partitionTaskExecutor) + .build(); +} +``` + +Full source: [`BatchConfig.java`](../src/main/java/com/ankurm/batchpartition/config/BatchConfig.java). +`StepBuilder.partitioner(String, Partitioner)` returns a `PartitionStepBuilder` +(`org.springframework.batch.core.step.builder.PartitionStepBuilder`, confirmed by decompiling +`spring-batch-core-6.0.5.jar` — see [chapter 3](03-what-gridsize-actually-controls.md)), +which is where `.step(...)`, `.taskExecutor(...)` and `.gridSize(...)` live. The `"ordersWorkerStep"` +string is the base name every partition's `StepExecution` is built from — it is why the logs +in [`docs/output/05-happy-path-4-partitions.txt`](output/05-happy-path-4-partitions.txt) show step +names like `ordersWorkerStep:partition0`. + +## Why there is no separately-coded "single-threaded baseline" + +The single most load-bearing design decision in this module: `--partition.grid-size=1` against a +one-shard directory drives the *exact same* `partitioner()` / `workerStep()` / `managerStep()` +code as `--partition.grid-size=8` against an eight-shard directory. There is no `if` branch, no +separate `@Profile`, no hand-written "just call the reader directly" fallback. The only two things +that change between every run in +[`docs/output/09-full-scale-throughput.txt`](output/09-full-scale-throughput.txt) are the number of +shard files on disk and the `gridSize`/pool-size properties passed on the command line. + +That matters because a partitioning benchmark that compares a hand-written single-threaded step +against a partitioned one is comparing two variables at once: thread count, AND whatever +incidental differences exist between the two code paths (a different reader class, a different +transaction boundary, different connection pool behavior). Isolating gridSize as the only variable +is what makes the numbers in [chapter 6](06-why-cpu-bound-not-io-bound.md) and +[chapter 10](10-scaling-sensitivity-to-data-size.md) mean what they claim to mean. + +## The worker step is nothing special + +```java +@Bean +public Step workerStep(JobRepository jobRepository, PlatformTransactionManager transactionManager, + FlatFileItemReader shardReader, ItemProcessor riskProcessor, + ItemWriter riskWriter, PartitionStatsListener partitionStatsListener) { + return new StepBuilder("ordersWorkerStep", jobRepository) + .chunk(1000) + .transactionManager(transactionManager) + .reader(shardReader) + .processor(riskProcessor) + .writer(riskWriter) + .listener(partitionStatsListener) + .build(); +} +``` + +Chunk size 1000, one processor ([chapter 6](06-why-cpu-bound-not-io-bound.md)), one writer +([chapter 4](04-the-writer-and-the-beanmapped-trap.md)). The only partition-aware piece is the +`@StepScope` reader, which late-binds to whichever file the manager step assigned it: + +```java +@Bean +@StepScope +public FlatFileItemReader shardReader(@Value("#{stepExecutionContext['fileName']}") Resource shardFile) { + return new FlatFileItemReaderBuilder() + .name("shardReader") + .resource(shardFile) + // ... + .build(); +} +``` + +`stepExecutionContext['fileName']` reads the exact key `MultiResourcePartitioner.setKeyName("fileName")` +wrote into each partition's `ExecutionContext` back in chapter 1's diagram. `@StepScope` is what +makes the late binding possible: this reader bean is not constructed until a specific +`StepExecution` (a specific partition) starts, at which point Spring resolves `#{...}` against +*that* execution's context, not some shared default. + +## Going deeper + +- `@StepScope` and late binding in general: + [Spring Batch reference — Late Binding](https://docs.spring.io/spring-batch/reference/step/late-binding.html) (`rel="nofollow"`). +- `PartitionStepBuilder`'s full decompiled surface: + [`docs/output/03-package-repackaging-javap.txt`](output/03-package-repackaging-javap.txt). + +[Next: What gridSize actually controls →](03-what-gridsize-actually-controls.md) diff --git a/spring-batch-partitioning/docs/03-what-gridsize-actually-controls.md b/spring-batch-partitioning/docs/03-what-gridsize-actually-controls.md new file mode 100644 index 0000000..c086227 --- /dev/null +++ b/spring-batch-partitioning/docs/03-what-gridsize-actually-controls.md @@ -0,0 +1,72 @@ +# 3. What gridSize actually controls + +[← Previous](02-anatomy-of-a-partitioned-step.md) | [README](../README.md) | [Next: The writer and the beanMapped trap →](04-the-writer-and-the-beanmapped-trap.md) + +Every tutorial on partitioning, including the reference documentation, describes `gridSize` as +"the number of partitions." That is true for the two built-in `Partitioner` implementations that +actually consult it (a custom range-based partitioner is expected to divide its input into +`gridSize` pieces). It is not true for `MultiResourcePartitioner`, the one this module uses, and +the gap between the two is easy to fall into. + +## Decompiling the claim + +`unzip -o spring-batch-core-6.0.5.jar org/springframework/batch/core/partition/support/MultiResourcePartitioner.class`, +then `javap -c` on it, shows `partition(int)` looping over the configured `resources` array and +never once loading its `int` parameter. The bytecode has no `iload_1` on the gridSize slot inside +the loop at all — only on the array-bounds check `iload; iload; if_icmpge`, which compares +the loop counter against `resources.length`, not against gridSize. + +## Proving it by running it, not just reading it + +[`PartitionerGridSizeTest`](../src/test/java/com/ankurm/batchpartition/PartitionerGridSizeTest.java) +asserts this directly: three real temp files, `partition(10)`, three partitions back: + +```console +$ mvn test -Dtest=PartitionerGridSizeTest +``` +Full transcript: [`docs/output/02-gridsize-ignored.txt`](output/02-gridsize-ignored.txt). + +And the same thing at the level of a real job: three shard files on disk, `--partition.grid-size=10`, +`--partition.pool-core-size=10`: + +```console +2026-09-14T09:10:38.720Z ... Executing step: [ordersWorkerStep:partition0] +2026-09-14T09:10:38.728Z ... Executing step: [ordersWorkerStep:partition2] +2026-09-14T09:10:38.732Z ... Executing step: [ordersWorkerStep:partition1] +JOB FINISHED: id=1 status=COMPLETED exitCode=COMPLETED +``` + +Three "Executing step" lines. Never a fourth, never a tenth, regardless of what `gridSize` says. + +## So what does gridSize control? + +Two things, both real, neither of them "how many partitions run" when your `Partitioner` ignores +the argument: + +1. **What gets passed to `Partitioner.partition(int)`.** A partitioner that *does* read its + argument (a hand-written range partitioner dividing one large table into `gridSize` key + ranges, for instance) is controlled by this value directly. `MultiResourcePartitioner` simply + happens not to be one of those. +2. **The `PartitionHandler`'s own accounting**, if you build one yourself with + `.partitionHandler(...)` instead of letting `PartitionStepBuilder` construct a default + `TaskExecutorPartitionHandler` from `.taskExecutor(...)` and `.gridSize(...)`. This module uses + the builder's default wiring, so its `gridSize` and `TaskExecutorPartitionHandler`'s internal + grid size are the same number by construction — but nothing stops them from diverging if + you wire a `PartitionHandler` bean explicitly with its own `setGridSize(...)`. + +The number of partitions that actually run is decided entirely by what +`Partitioner.partition(gridSize)` **returns** — a `Map` — not by the `int` it was +handed. For `MultiResourcePartitioner`, that means: however many files are in +`partition.shards-dir`. Full stop. Sizing `gridSize` to match core count (chapter 6) only works if +you also size the number of shard files to match, which this module's benchmarking scripts do +deliberately (see [`scripts/generate-shards.py`](../scripts/generate-shards.py)). + +## Going deeper + +- `Partitioner` and the other built-in implementation, `SimplePartitioner` (one partition, no + data division at all — used internally when no explicit partitioner is set): + [Spring Batch reference — the Partitioner interface](https://docs.spring.io/spring-batch/reference/scalability.html#partitioner-interface) (`rel="nofollow"`). +- Writing a partitioner that *does* use gridSize (a key-range partitioner over a database table): + [Spring Batch samples — ColumnRangePartitioner](https://github.com/spring-projects/spring-batch/tree/main/spring-batch-samples/src/main/java/org/springframework/batch/samples/partitioning) (`rel="nofollow"`). + +[Next: The writer and the beanMapped trap →](04-the-writer-and-the-beanmapped-trap.md) diff --git a/spring-batch-partitioning/docs/04-the-writer-and-the-beanmapped-trap.md b/spring-batch-partitioning/docs/04-the-writer-and-the-beanmapped-trap.md new file mode 100644 index 0000000..70ea80c --- /dev/null +++ b/spring-batch-partitioning/docs/04-the-writer-and-the-beanmapped-trap.md @@ -0,0 +1,72 @@ +# 4. The writer, the beanMapped trap, and finding the partition's own name + +[← Previous](03-what-gridsize-actually-controls.md) | [README](../README.md) | [Next: The diagnostic endpoint →](05-the-diagnostic-endpoint.md) + +## Not beanMapped(), again + +[The earlier `spring-batch` module](https://ankurm.com/) already found this once: `Order` and +`RiskScoredOrder` are Java records, and `JdbcBatchItemWriterBuilder.beanMapped()` binds SQL +parameters through `BeanPropertySqlParameterSource`, which looks for JavaBean-style getters +(`getOrderId()`). A record's accessors are `orderId()`, no `get` prefix — `beanMapped()` +would silently bind every column to `NULL` against this exact shape, because +`BeanPropertySqlParameterSource` treats a missing getter as an absent property rather than an +error. This module's [`riskWriter`](../src/main/java/com/ankurm/batchpartition/config/BatchConfig.java) +uses an explicit `itemPreparedStatementSetter` instead, precisely to avoid rediscovering that bug +a second time. + +## Finding out which partition wrote which row + +`ORDER_RISK_SUMMARY.partition_name` exists so a reader (or a bug report) can tell which worker +wrote a given row. The first attempt to populate it reached for +`StepSynchronizationManager.getContext().getStepExecution().getStepName()` inside the +`itemPreparedStatementSetter` lambda, guessing the class lived at +`org.springframework.batch.core.step.StepSynchronizationManager` by analogy with `StepExecution` +living in `org.springframework.batch.core.step`. It does not compile: + +``` +error: package org.springframework.batch.core.step does not exist + (well -- StepSynchronizationManager specifically does not exist there) +``` + +`javap` against the real jar answers where it actually is: +[`docs/output/03-package-repackaging-javap.txt`](output/03-package-repackaging-javap.txt) shows +`org.springframework.batch.core.scope.context.StepSynchronizationManager` — alongside +`JobScope` and `StepScope`'s own scope-context machinery, which is the more sensible package for +it in hindsight; the guess just followed the wrong analogy. + +The fix that shipped is simpler than getting the import right would have been: late-bind the step +name directly, the same mechanism [chapter 2](02-anatomy-of-a-partitioned-step.md) already uses +for the shard file path. + +```java +@Bean +@StepScope +public ItemWriter riskWriter(JdbcTemplate jdbcTemplate, + @Value("#{stepExecution.stepName}") String partitionName) { + return new JdbcBatchItemWriterBuilder() + .dataSource(jdbcTemplate.getDataSource()) + .sql("MERGE INTO ORDER_RISK_SUMMARY (order_id, customer_id, amount_cents, region, " + + "risk_score, high_risk, partition_name) KEY(order_id) VALUES (?, ?, ?, ?, ?, ?, ?)") + .itemPreparedStatementSetter((item, ps) -> { + // ... ps.setString(7, partitionName); + }) + .assertUpdates(true) + .build(); +} +``` + +`#{stepExecution.stepName}` reaches the *current* `StepExecution`'s name (`ordersWorkerStep:partition2`, +for instance) through the same `@StepScope` proxy machinery that resolves `#{stepExecutionContext['fileName']}` +— no lookup class needed, no package to get wrong. `MERGE ... KEY(order_id)` (H2 upsert +syntax) rather than a plain `INSERT` was a deliberate second choice: it makes re-running a single +partition idempotent, which matters once [chapter 8](08-restart-reruns-only-the-failed-partition.md) +starts re-executing partitions after a failure. + +## Going deeper + +- `BeanPropertySqlParameterSource` and why it fails silently rather than throwing: + [Spring Framework Javadoc](https://docs.spring.io/spring-framework/docs/current/javadoc-api/org/springframework/jdbc/core/BeanPropertySqlParameterSource.html) (`rel="nofollow"`). +- The full list of classes this article's research found relocated in Spring Batch 6.0: + [`docs/output/03-package-repackaging-javap.txt`](output/03-package-repackaging-javap.txt). + +[Next: The diagnostic endpoint →](05-the-diagnostic-endpoint.md) diff --git a/spring-batch-partitioning/docs/05-the-diagnostic-endpoint.md b/spring-batch-partitioning/docs/05-the-diagnostic-endpoint.md new file mode 100644 index 0000000..3de0c5a --- /dev/null +++ b/spring-batch-partitioning/docs/05-the-diagnostic-endpoint.md @@ -0,0 +1,64 @@ +# 5. The diagnostic endpoint: proving four threads ran, not just configuring them + +[← Previous](04-the-writer-and-the-beanmapped-trap.md) | [README](../README.md) | [Next: Why CPU-bound, not I/O-bound →](06-why-cpu-bound-not-io-bound.md) + +Configuration says how many partitions *should* run and on how many threads. It does not say what +actually happened on a given execution — whether the pool was saturated, whether two +partitions landed on the same thread, how long each one actually took relative to the others. This +module makes that observable instead of assumed. + +## PartitionStatsListener + +[`PartitionStatsListener`](../src/main/java/com/ankurm/batchpartition/partition/PartitionStatsListener.java) +is a `StepExecutionListener` attached to the worker step. `beforeStep` records a start timestamp on +a `ThreadLocal` (each partition runs on its own thread, so there is no cross-talk); `afterStep` +computes the duration and inserts one row into `PARTITION_STATS`: which step (which partition), +which OS thread name, how many rows it read, when it started and finished, how long it took, and +its exit code. + +## PartitionInsightController + +[`PartitionInsightController`](../src/main/java/com/ankurm/batchpartition/web/PartitionInsightController.java) +exposes that table over HTTP: + +```console +$ curl -s http://localhost:8081/batch/partitions/1 | python3 -m json.tool +[ + { + "PARTITION_NAME": "ordersWorkerStep:partition0", + "THREAD_NAME": "order-partition-2", + "READ_COUNT": 5000, + "STARTED_AT": "2026-09-14T09:29:56.892Z", + "FINISHED_AT": "2026-09-14T09:29:57.968Z", + "DURATION_MS": 1075, + "EXIT_CODE": "COMPLETED" + }, + ... +] +``` + +Full transcript, all four partitions: [`docs/output/05-happy-path-4-partitions.txt`](output/05-happy-path-4-partitions.txt). +Two things this makes concrete that configuration alone does not: partition-to-thread assignment +is not 1:1 with partition *number* (`partition0` landed on thread `order-partition-2` here, not +`order-partition-1` — `TaskExecutorPartitionHandler` submits tasks in the order the +`Partitioner`'s map iterates, and `ThreadPoolTaskExecutor` hands them to whichever pooled thread is +free first); and all four really did start within milliseconds of each other, which is the +difference between "partitioned" as a configuration property and "partitioned" as an observed +fact. + +This same table is what makes [chapter 7](07-the-rejectedexecutionexception.md) and +[chapter 8](08-restart-reruns-only-the-failed-partition.md) possible to write with confidence: +querying `PARTITION_STATS` (or, for those chapters, the framework's own `BATCH_STEP_EXECUTION` +table directly) is how this article found that three rejected partitions do not fail, they hang. + +**Delete this before shipping.** An unauthenticated endpoint that dumps job-internal timing data +belongs behind whatever authentication and authorization the rest of the service already has, at +minimum, and arguably behind nothing publicly reachable at all — it exists here to make the +mechanism visible for this article, not because a production batch service should expose it. + +## Going deeper + +- `StepExecutionListener` and the other listener interfaces available: + [Spring Batch reference — Listeners](https://docs.spring.io/spring-batch/reference/step/chunk-oriented-processing/configuring-a-step.html) (`rel="nofollow"`). + +[Next: Why CPU-bound, not I/O-bound →](06-why-cpu-bound-not-io-bound.md) diff --git a/spring-batch-partitioning/docs/06-why-cpu-bound-not-io-bound.md b/spring-batch-partitioning/docs/06-why-cpu-bound-not-io-bound.md new file mode 100644 index 0000000..3345bb6 --- /dev/null +++ b/spring-batch-partitioning/docs/06-why-cpu-bound-not-io-bound.md @@ -0,0 +1,61 @@ +# 6. Why this module's work is CPU-bound, not I/O-bound + +[← Previous](05-the-diagnostic-endpoint.md) | [README](../README.md) | [Next: The RejectedExecutionException →](07-the-rejectedexecutionexception.md) + +## The design choice + +[`RiskScoringProcessor`](../src/main/java/com/ankurm/batchpartition/processing/RiskScoringProcessor.java) +does `risk.iterations` (150 by default) real 64-bit XOR/multiply operations per order, then flags +the order high-risk if the resulting score exceeds a threshold or the amount exceeds +`risk.high-risk-amount-cents`. No network call, no sleep, no artificial delay — every +millisecond it spends is spent computing, on whichever core its thread is scheduled on. + +That is deliberate. A partitioning demo built around an I/O-bound step (a network call per item, a +slow downstream service) shows a speedup even on a single-core machine, because the threads spend +almost all their time blocked, not competing for CPU — the "parallelism" it demonstrates is +really just concurrency hiding latency, which is a real and useful thing but a different claim than +"this uses more of the machine's compute." A CPU-bound processor makes the speedup measured in +[`docs/output/09-full-scale-throughput.txt`](output/09-full-scale-throughput.txt) mean what it +looks like it means: more cores actually doing more arithmetic per second, capped by how many +cores physically exist. + +## Determinism, and what it buys + +[`RiskScoringProcessorTest`](../src/test/java/com/ankurm/batchpartition/processing/RiskScoringProcessorTest.java) +pins that the same `Order` always produces the same score: + +```console +order: Order[orderId=42, customerId=777, amountCents=1234567, region=NORTH] +first.process() -> riskScore=61 highRisk=false +second.process() -> riskScore=61 highRisk=false +``` +Full transcript: [`docs/output/01-processor-determinism.txt`](output/01-processor-determinism.txt). + +Determinism matters for a partitioned job specifically because of +[chapter 8](08-restart-reruns-only-the-failed-partition.md): a partition that fails and re-runs +must produce the same output the second time, or a restart silently changes results depending on +which attempt happened to write. `MERGE ... KEY(order_id)` in the writer +([chapter 4](04-the-writer-and-the-beanmapped-trap.md)) handles the "don't duplicate the row" +half of that; a deterministic processor handles the "don't change the row's *content* between +attempts" half. + +## What the CPU cost does and does not explain about the scaling numbers + +Raising `risk.iterations` to 5000 (thirty-three times the default) at a fixed 300,000-row data +volume moved the grid-size-2 speedup from 1.10x to 1.26x — see +[`docs/output/09-full-scale-throughput.txt`](output/09-full-scale-throughput.txt) for both numbers +side by side. That is evidence that at least part of the shortfall from a clean 2x speedup on 2 +cores is time spent somewhere that does *not* scale with thread count — the shared H2 +writer is the leading candidate, discussed further in +[chapter 10](10-scaling-sensitivity-to-data-size.md) — but it is not proof by itself; this +module did not isolate the writer completely (by, say, writing to per-partition tables and +comparing) to rule out GC pressure or thread scheduling overhead as contributing causes too. Read +the 1.10x-to-1.26x shift as "consistent with the write-contention hypothesis," not as a closed +case. + +## Going deeper + +- Amdahl's law, for the general shape of "some fraction of the work cannot be parallelized": + [Wikipedia — Amdahl's law](https://en.wikipedia.org/wiki/Amdahl%27s_law) (`rel="nofollow"`). + +[Next: The RejectedExecutionException →](07-the-rejectedexecutionexception.md) diff --git a/spring-batch-partitioning/docs/07-the-rejectedexecutionexception.md b/spring-batch-partitioning/docs/07-the-rejectedexecutionexception.md new file mode 100644 index 0000000..dade8e7 --- /dev/null +++ b/spring-batch-partitioning/docs/07-the-rejectedexecutionexception.md @@ -0,0 +1,82 @@ +# 7. The failure that does not look like a failure: rejected partitions + +[← Previous](06-why-cpu-bound-not-io-bound.md) | [README](../README.md) | [Next: Restart reruns only the failed partition →](08-restart-reruns-only-the-failed-partition.md) + +`TaskExecutorPartitionHandler` submits every partition's worker step to its `TaskExecutor` up +front, then waits for all of them. What happens if the executor cannot accept all of those +submissions? + +## Reproducing it + +The `reject` Spring profile swaps in a deliberately undersized `ThreadPoolTaskExecutor`: +`corePoolSize`/`maxPoolSize` both 1, `queueCapacity` 0, and +`java.util.concurrent.ThreadPoolExecutor.AbortPolicy` as the rejection handler (Spring's own +default, `CallerRunsPolicy`, would instead silently run the "extra" partitions on the submitting +thread one at a time — serializing them rather than rejecting them, which hides this problem +instead of surfacing it). Four shard files, pool size one: + +```console +$ java -jar target/spring-batch-partitioning-1.0.0.jar --spring.profiles.active=reject \ + --partition.shards-dir=./data/shards-small --partition.grid-size=4 --partition.reject.pool-size=1 +``` + +Only one partition ever logs "Executing step." The job fails with +`JobExecutionException: Partition handler returned an unsuccessful step` — a message that +never mentions rejection, threads, or pools. Full transcript: +[`docs/output/06-rejected-partitions-stuck.txt`](output/06-rejected-partitions-stuck.txt). + +## What actually happened, found by querying the job repository directly + +Nothing in the console log says `TaskRejectedException` anywhere. +`TaskExecutorPartitionHandler.doHandle` catches whatever the executor throws on submission and +folds it into the corresponding `StepExecution`'s failure exceptions rather than logging it, so +the only way to see it is to look at the repository's own tables: + +```console +$ java -cp h2-2.4.240.jar org.h2.tools.Shell -url jdbc:h2:file:./data/rejecttest -user sa -password "" \ + -sql "SELECT STEP_EXECUTION_ID, STEP_NAME, STATUS, EXIT_CODE FROM BATCH_STEP_EXECUTION ORDER BY STEP_EXECUTION_ID;" + +STEP_EXECUTION_ID | STEP_NAME | STATUS | EXIT_CODE +1 | ordersManagerStep | FAILED | FAILED +2 | ordersWorkerStep:partition3 | STARTING | EXECUTING +3 | ordersWorkerStep:partition2 | STARTING | EXECUTING +4 | ordersWorkerStep:partition1 | COMPLETED | COMPLETED +5 | ordersWorkerStep:partition0 | STARTING | EXECUTING +``` + +The manager step and the `JobExecution` both reach a clean `FAILED`. The three rejected worker +`StepExecution`s do not. They are permanently parked at `STARTING`/`EXECUTING` — not failed, +not running, just stuck — because nothing ever calls back into the job repository to close +them out; the executor rejected them before Spring Batch's own bookkeeping around that +`StepExecution` ever started. + +
+ + +partition1runs, COMPLETED +partition0rejected on submit +partition2rejected on submit +partition3rejected on submit +StepExecution status for the three rejected partitions: + +STARTING / EXECUTING -- forever. No callback ever closes them out. +The manager step and JobExecution both reach FAILED cleanly. These three do not. + +
+ +## Why this is worse than an ordinary failure + +An ordinary failed step is exactly what restart exists for. This is not that — see +[chapter 8](08-restart-reruns-only-the-failed-partition.md) for what happens when you try to +restart a job in this state, and [chapter 9](09-jobexecutionalreadyrunning-and-recover.md) for the +Spring Batch 6.0 feature that exists specifically to get out of it. + +## Going deeper + +- `ThreadPoolTaskExecutor`'s rejection handler options (`AbortPolicy`, `CallerRunsPolicy`, + `DiscardPolicy`, `DiscardOldestPolicy`) and their very different failure characters: + [`ThreadPoolExecutor` Javadoc](https://docs.spring.io/spring-framework/docs/current/javadoc-api/org/springframework/scheduling/concurrent/ThreadPoolTaskExecutor.html) (`rel="nofollow"`). +- Sizing a pool correctly relative to gridSize in the first place: + [chapter 3](03-what-gridsize-actually-controls.md). + +[Next: Restart reruns only the failed partition →](08-restart-reruns-only-the-failed-partition.md) diff --git a/spring-batch-partitioning/docs/08-restart-reruns-only-the-failed-partition.md b/spring-batch-partitioning/docs/08-restart-reruns-only-the-failed-partition.md new file mode 100644 index 0000000..45bcc8c --- /dev/null +++ b/spring-batch-partitioning/docs/08-restart-reruns-only-the-failed-partition.md @@ -0,0 +1,74 @@ +# 8. Restart reruns only the failed partition — proved, not assumed + +[← Previous](07-the-rejectedexecutionexception.md) | [README](../README.md) | [Next: JobExecutionAlreadyRunningException, and recover →](09-jobexecutionalreadyrunning-and-recover.md) + +The reference documentation states that a restarted partitioned step only re-executes partitions +that did not complete. This chapter is that claim, checked against a real `BATCH_STEP_EXECUTION` +table rather than taken on faith — using exactly the stuck job from +[chapter 7](07-the-rejectedexecutionexception.md), after +[chapter 9](09-jobexecutionalreadyrunning-and-recover.md)'s recovery step unsticks it. + +## The evidence + +```console +$ java -cp h2-2.4.240.jar org.h2.tools.Shell -url jdbc:h2:file:./data/rejecttest -user sa -password "" \ + -sql "SELECT JOB_EXECUTION_ID, STEP_EXECUTION_ID, STEP_NAME, STATUS, READ_COUNT + FROM BATCH_STEP_EXECUTION WHERE JOB_EXECUTION_ID IN (1,33) ORDER BY JOB_EXECUTION_ID, STEP_EXECUTION_ID;" + +JOB_EXECUTION_ID | STEP_EXECUTION_ID | STEP_NAME | STATUS | READ_COUNT +1 | 1 | ordersManagerStep | FAILED | 5000 +1 | 2 | ordersWorkerStep:partition3 | FAILED | 0 +1 | 3 | ordersWorkerStep:partition2 | FAILED | 0 +1 | 4 | ordersWorkerStep:partition1 | COMPLETED | 5000 +1 | 5 | ordersWorkerStep:partition0 | FAILED | 0 +33 | 33 | ordersManagerStep | COMPLETED | 15000 +33 | 34 | ordersWorkerStep:partition3 | COMPLETED | 5000 +33 | 35 | ordersWorkerStep:partition2 | COMPLETED | 5000 +33 | 36 | ordersWorkerStep:partition0 | COMPLETED | 5000 +33 | 37 | reportStep | COMPLETED | 0 +``` + +Full transcript: [`docs/output/08-recover-then-restart.txt`](output/08-recover-then-restart.txt). +`JobExecution` 33 — the restart — has exactly three new worker `StepExecution` rows: +`partition3`, `partition2`, `partition0`, the three [chapter 9](09-jobexecutionalreadyrunning-and-recover.md)'s +recovery step marked `FAILED`. There is no new row for `partition1`. It stayed `COMPLETED` from +`JobExecution` 1, at `READ_COUNT` 5000, untouched. `ordersManagerStep`'s own `READ_COUNT` for +execution 33 is 15000 — the sum of the three re-run partitions, not all four. + +
+ + +JobExecution 1 +partition1: COMPLETED +partition0: FAILED +partition2: FAILED +partition3: FAILED +JobExecution 33 (restart, same shardsDir) +partition1: SKIPPED +partition0: reran +partition2: reran +partition3: reran +manager READ_COUNT for execution 33 = 15000 (3 x 5000), not 20000 + +
+ +## Why this works: partition names are stable across attempts + +`MultiResourcePartitioner` names partitions `partition0`, `partition1`, ... in the order its +`resources` array iterates ([chapter 3](03-what-gridsize-actually-controls.md)), and that order is +deterministic once the shard files and their sort order are fixed — `BatchConfig.partitioner()` +explicitly sorts `resources` by filename before handing them to `MultiResourcePartitioner` for +exactly this reason. Restart resolves each partition's `StepExecution` by that stable name against +the same `JobInstance`, the same mechanism [the earlier `spring-batch` module](https://ankurm.com/) +demonstrated for a single, unpartitioned step: same identifying job parameters, same instance, +already-`COMPLETED` steps skipped. Partitioning does not add a different restart mechanism; it +applies the same one once per partition. + +## Going deeper + +- Ordinary (unpartitioned) restart semantics, which this chapter's mechanism is not different + from: [Spring Batch on Boot 4.1, chapter 7](https://ankurm.com/). +- Why `JobExecution` 1 needed a recovery step before this restart was even possible: + [chapter 9](09-jobexecutionalreadyrunning-and-recover.md). + +[Next: JobExecutionAlreadyRunningException, and recover →](09-jobexecutionalreadyrunning-and-recover.md) diff --git a/spring-batch-partitioning/docs/09-jobexecutionalreadyrunning-and-recover.md b/spring-batch-partitioning/docs/09-jobexecutionalreadyrunning-and-recover.md new file mode 100644 index 0000000..9adbb77 --- /dev/null +++ b/spring-batch-partitioning/docs/09-jobexecutionalreadyrunning-and-recover.md @@ -0,0 +1,82 @@ +# 9. JobExecutionAlreadyRunningException, forever — and Spring Batch 6.0's `recover()` + +[← Previous](08-restart-reruns-only-the-failed-partition.md) | [README](../README.md) | [Next: Scaling sensitivity to data size →](10-scaling-sensitivity-to-data-size.md) + +[Chapter 7](07-the-rejectedexecutionexception.md) left `JobExecution` 1 with three worker +`StepExecution`s permanently parked at `STARTING`. This chapter is what trying to move on from +that actually looks like. + +## The job cannot be restarted + +The same command that would normally restart a failed job, run again against the same +`shardsDir`: + +```console +$ java -jar target/spring-batch-partitioning-1.0.0.jar \ + --partition.shards-dir=./data/shards-small --partition.grid-size=4 --partition.pool-core-size=4 --partition.pool-max-size=4 +``` + +throws, every single time: + +``` +Caused by: org.springframework.batch.core.launch.JobExecutionAlreadyRunningException: A job execution +for this job is already running: JobExecution: id=1, version=3, ... status=FAILED, exitStatus=exitCode=FAILED;... +``` + +Full transcript: [`docs/output/07-restart-throws-alreadyrunning.txt`](output/07-restart-throws-alreadyrunning.txt). +Read the exception's own embedded string closely: it says `status=FAILED` in the same message that +claims the execution is "already running." `SimpleJobOperator`'s running-check is not "is the +`JobExecution` status `FAILED`" — it is closer to "does this `JobInstance` have any +`StepExecution` that has not reached a terminal status," and the three orphaned +`STARTING`/`EXECUTING` worker steps from chapter 7 satisfy that condition indefinitely. There is no +number of retries that fixes this on its own. The job is not failed. It is stuck. + +## `JobOperator#recover`, new in Spring Batch 6.0 + +6.0 added exactly the operation this situation calls for: +`JobOperator.recover(JobExecution)`. [`RecoveryRunner`](../src/main/java/com/ankurm/batchpartition/runner/RecoveryRunner.java) +(active under the `recover` Spring profile, `@Order(0)` so it runs before the normal launch logic) +fetches the stuck execution via `JobExplorer` and calls it: + +```console +$ java -jar target/spring-batch-partitioning-1.0.0.jar --spring.profiles.active=recover \ + --recover.job-execution-id=1 --partition.shards-dir=./data/shards-small \ + --partition.grid-size=4 --partition.pool-core-size=4 --partition.pool-max-size=4 + +RECOVER: before -> status=FAILED +RECOVER: step=ordersWorkerStep:partition3 status=STARTING +RECOVER: step=ordersWorkerStep:partition2 status=STARTING +RECOVER: step=ordersWorkerStep:partition1 status=COMPLETED +RECOVER: step=ordersWorkerStep:partition0 status=STARTING +RECOVER: after -> status=FAILED +RECOVER: step=ordersWorkerStep:partition3 status=FAILED +RECOVER: step=ordersWorkerStep:partition2 status=FAILED +RECOVER: step=ordersWorkerStep:partition1 status=COMPLETED +RECOVER: step=ordersWorkerStep:partition0 status=FAILED +JOB FINISHED: id=33 status=COMPLETED exitCode=COMPLETED +``` + +Full transcript: [`docs/output/08-recover-then-restart.txt`](output/08-recover-then-restart.txt). +`recover()` force-closes every non-terminal `StepExecution` under the given `JobExecution` to +`FAILED` — nothing else. The already-`COMPLETED` `partition1` is left exactly as it was. +Immediately afterward, in the same JVM, `OrderIngestRunner`'s ordinary `jobOperator.start(...)` +call runs against the same `shardsDir` and this time succeeds, producing `JobExecution` 33 — +which [chapter 8](08-restart-reruns-only-the-failed-partition.md) shows re-executed only the three +partitions `recover()` had just marked `FAILED`. + +## The takeaway + +A `RejectedExecutionException` (or any other exception that reaches the executor *before* a +`StepExecution` starts, rather than during it) is a different failure class than an ordinary step +failure, and it needs a different remedy: `recover()`, then restart. Sizing the pool correctly in +the first place ([chapter 3](03-what-gridsize-actually-controls.md)) avoids the situation +entirely; `recover()` is what to reach for when a job in production has already gotten into it. + +## Going deeper + +- `JobOperator`'s full interface, including `recover`, `abandon`, and `stop`: + [`docs/output/03-package-repackaging-javap.txt`](output/03-package-repackaging-javap.txt) has the + decompiled method list this chapter's claims were checked against. +- The stuck state this chapter recovers from: [chapter 7](07-the-rejectedexecutionexception.md). + +[Next: Scaling sensitivity to data size →](10-scaling-sensitivity-to-data-size.md) diff --git a/spring-batch-partitioning/docs/10-scaling-sensitivity-to-data-size.md b/spring-batch-partitioning/docs/10-scaling-sensitivity-to-data-size.md new file mode 100644 index 0000000..c5d1e36 --- /dev/null +++ b/spring-batch-partitioning/docs/10-scaling-sensitivity-to-data-size.md @@ -0,0 +1,82 @@ +# 10. Scaling sensitivity to data size, and the honest ceiling + +[← Previous](09-jobexecutionalreadyrunning-and-recover.md) | [README](../README.md) | [Next: Production checklist →](11-production-checklist.md) + +## The full-scale numbers + +Same job, same code path ([chapter 2](02-anatomy-of-a-partitioned-step.md)), same deterministically +seeded 10,000,000-row dataset, only `partition.grid-size` and the shard directory changing, on this +module's 2-vCPU sandbox: + +| grid-size | manager-step wall time | rows/sec | speedup vs. grid-size 1 | +|---|---|---|---| +| 1 | 70.485s | 141,874 | 1.00x | +| 2 | 49.647s | 201,422 | **1.42x (best)** | +| 4 | 53.102s | 188,317 | 1.33x | +| 8 | 55.521s | 180,112 | 1.27x | + +Full transcript: [`docs/output/09-full-scale-throughput.txt`](output/09-full-scale-throughput.txt). +Best result at grid-size 2 — exactly the physical core count. Past that, wall time gets +*worse* with every doubling: more partitions than cores does not sit still, it costs real time in +scheduling and context-switch overhead with no additional compute to absorb it. + +## The same sweep at 300,000 rows tells a different story + +| grid-size | wall time | speedup | +|---|---|---| +| 1 | 5.127s | 1.00x | +| 2 | 4.677s | 1.10x | +| 4 | 5.158s | 0.99x | +| 8 | 6.812s | **0.75x (worse than not partitioning at all)** | + +At the smaller volume, grid-size 8 does not just lose to grid-size 2 — it loses to grid-size +1. The fixed cost of standing up one partition (opening its shard file, acquiring a JDBC +connection from the pool, a thread handoff) is roughly the same fixed number of milliseconds +whichever scale the job runs at. At 10,000,000 rows there is enough real work per partition to +amortize that fixed cost into irrelevance; at 300,000 rows split eight ways (37,500 rows each) +there is not. **Over-partitioning is a strictly worse mistake on a smaller job than on a larger +one** — the "right" gridSize is a function of data volume as well as core count, not core +count alone, and a gridSize tuned against a large nightly batch can be actively harmful applied +unchanged to a smaller one. + +
+ + + + +speedup +gridSize +1.0x + + + + +10M rows + + + +300K rows +1248 + +
+ +## The two candidate explanations for the sub-2x ceiling at the best setting + +Even at the best-measured setting (grid-size 2 on 2 cores), speedup tops out at 1.42x, not 2x. +[Chapter 6](06-why-cpu-bound-not-io-bound.md) found one piece of evidence pointing at the shared +H2 file database as a contributing serialization point: raising per-item CPU cost (5000 iterations +instead of 150, at the 300K scale) moved the grid-size-2 speedup from 1.10x to 1.26x, consistent +with diluting a fixed write-lock cost across more total work — but this module did not +isolate the writer completely (per-partition tables, or an in-memory sink, compared directly), so +treat that as evidence pointing in a direction, not a closed investigation. A production job +choosing a real target database (one built for concurrent writers, not a single embedded MVStore +file) would need to re-measure this, not assume the same ceiling applies. + +## Going deeper + +- H2's MVStore concurrency model (single writer per store): + [H2 MVStore documentation](https://h2database.com/html/mvstore.html) (`rel="nofollow"`). +- The CPU-vs-I/O-bound design choice this chapter's numbers depend on: + [chapter 6](06-why-cpu-bound-not-io-bound.md). + +[Next: Production checklist →](11-production-checklist.md) diff --git a/spring-batch-partitioning/docs/11-production-checklist.md b/spring-batch-partitioning/docs/11-production-checklist.md new file mode 100644 index 0000000..000cdec --- /dev/null +++ b/spring-batch-partitioning/docs/11-production-checklist.md @@ -0,0 +1,51 @@ +# 11. Production checklist + +[← Previous](10-scaling-sensitivity-to-data-size.md) | [README](../README.md) + +Everything below is a consequence of an earlier chapter, not a new claim — this is the +condensed version to check a real job against. + +- **Confirm your `Partitioner` actually reads `gridSize` before sizing anything by it.** + `MultiResourcePartitioner` does not ([chapter 3](03-what-gridsize-actually-controls.md)). The + number of partitions that run is whatever your partitioner's `partition()` map returns, full + stop — check that, not the `gridSize` property, when something needs exactly N partitions. + +- **Size the thread pool to at least gridSize, with a queue, not a bare `AbortPolicy` at zero + capacity.** An undersized pool with `AbortPolicy` does not fail loudly — it leaves + rejected partitions' `StepExecution`s parked at `STARTING` forever and makes every future + restart throw `JobExecutionAlreadyRunningException` ([chapter 7](07-the-rejectedexecutionexception.md), + [chapter 9](09-jobexecutionalreadyrunning-and-recover.md)). If you must use `AbortPolicy` for + fail-fast behavior, have an operational runbook that calls `JobOperator#recover` before anyone + tries to restart. + +- **Benchmark gridSize against your actual data volume, not just your core count.** More + partitions than cores got *worse*, not flat, at both scales this module measured; at the + smaller of the two, over-partitioning lost to not partitioning at all + ([chapter 10](10-scaling-sensitivity-to-data-size.md)). A gridSize tuned for a large nightly + batch is not automatically safe for a smaller one. + +- **Check what your worker steps write to.** A single-writer embedded database (this module's H2 + file) caps the speedup partitioning can deliver regardless of thread count or core count + ([chapter 6](06-why-cpu-bound-not-io-bound.md), [chapter 10](10-scaling-sensitivity-to-data-size.md)). + A database built for concurrent writers changes this ceiling; measure again against the real + target rather than assuming. + +- **Make the writer idempotent if partitions can retry.** `MERGE ... KEY(order_id)` here, not + `INSERT` ([chapter 4](04-the-writer-and-the-beanmapped-trap.md)) — a partition that fails + after partially writing and then restarts writes some rows twice on a plain `INSERT`. + +- **Delete the diagnostic endpoint** ([chapter 5](05-the-diagnostic-endpoint.md)) or put it behind + real authentication before this leaves a sandbox. + +- **Don't reach for `BeanPropertySqlParameterSource`/`beanMapped()` with Java records** without + checking it actually populated your columns ([chapter 4](04-the-writer-and-the-beanmapped-trap.md)) + — it fails by silently writing `NULL`, not by throwing. + +- **If migrating a 5.x partitioned job to 6.0**, budget time for the `org.springframework.batch.item.*` + → `org.springframework.batch.infrastructure.item.*` import rewrite across every reader, + writer, and `ExecutionContext` reference before anything else compiles + ([`docs/output/03-package-repackaging-javap.txt`](output/03-package-repackaging-javap.txt)), and + rename any `@Bean` methods that were only distinguished by `@Profile` + ([`docs/output/04-enforceuniquemethods-error.txt`](output/04-enforceuniquemethods-error.txt)). + +[← Previous](10-scaling-sensitivity-to-data-size.md) | [README](../README.md) diff --git a/spring-batch-partitioning/docs/output/01-processor-determinism.txt b/spring-batch-partitioning/docs/output/01-processor-determinism.txt new file mode 100644 index 0000000..1a45a1c --- /dev/null +++ b/spring-batch-partitioning/docs/output/01-processor-determinism.txt @@ -0,0 +1,9 @@ +# RiskScoringProcessor determinism, threshold behaviour + +order: Order[orderId=42, customerId=777, amountCents=1234567, region=NORTH] +first.process() -> riskScore=61 highRisk=false +second.process() -> riskScore=61 highRisk=false + +--- amount threshold, score held constant by a low iteration count --- +amountCents=9,499,999 (at threshold, exclusive) -> highRisk=false +amountCents=9,500,001 (over threshold) -> highRisk=true diff --git a/spring-batch-partitioning/docs/output/02-gridsize-ignored.txt b/spring-batch-partitioning/docs/output/02-gridsize-ignored.txt new file mode 100644 index 0000000..8c15f39 --- /dev/null +++ b/spring-batch-partitioning/docs/output/02-gridsize-ignored.txt @@ -0,0 +1,6 @@ +# MultiResourcePartitioner.partition(10) with 3 resources + +resources given: 3 +gridSize argument passed to partition(): 10 +partitions actually returned: 3 +partition keys: partition2, partition1, partition0 diff --git a/spring-batch-partitioning/docs/output/03-package-repackaging-javap.txt b/spring-batch-partitioning/docs/output/03-package-repackaging-javap.txt new file mode 100644 index 0000000..83f9017 --- /dev/null +++ b/spring-batch-partitioning/docs/output/03-package-repackaging-javap.txt @@ -0,0 +1,43 @@ +# Spring Batch 6.0: the item/repeat infrastructure moved packages + +Verified by decompiling the real jars pulled from Maven Central, not by reading prose. + +--- Spring Batch 5.2.x (docs.spring.io javadoc, org.springframework.batch:spring-batch-infrastructure:5.2.6) --- +Package: org.springframework.batch.item +Class: org.springframework.batch.item.ExecutionContext + +--- Spring Batch 6.0.5 (unzip -l spring-batch-infrastructure-6.0.5.jar) --- +$ unzip -l spring-batch-infrastructure-6.0.5.jar | grep -E "ExecutionContext.class|CompletionPolicy.class" + 2908 2026-08-17 10:28 org/springframework/batch/infrastructure/repeat/policy/SimpleCompletionPolicy.class + 3388 2026-08-17 10:28 org/springframework/batch/infrastructure/repeat/policy/CompositeCompletionPolicy.class + 1179 2026-08-17 10:28 org/springframework/batch/infrastructure/repeat/policy/DefaultResultCompletionPolicy.class + 2934 2026-08-17 10:28 org/springframework/batch/infrastructure/repeat/policy/CountingCompletionPolicy.class + 718 2026-08-17 10:28 org/springframework/batch/infrastructure/repeat/CompletionPolicy.class + 7922 2026-08-17 10:28 org/springframework/batch/infrastructure/item/ExecutionContext.class + +Every class under the old org.springframework.batch.item.* and org.springframework.batch.repeat.* +now lives under org.springframework.batch.infrastructure.item.* and +org.springframework.batch.infrastructure.repeat.* -- ItemReader, ItemWriter, ItemProcessor, +ExecutionContext, RepeatStatus, CompletionPolicy, every FlatFileItemReaderBuilder and +JdbcBatchItemWriterBuilder, all of it. A 5.x guide's imports do not compile against 6.0 for this +reason alone, before any API shape has changed at all. + +--- Partitioner interface, decompiled from spring-batch-core-6.0.5.jar --- +$ javap org/springframework/batch/core/partition/Partitioner.class +Compiled from "Partitioner.java" +public interface org.springframework.batch.core.partition.Partitioner { + public abstract java.util.Map partition(int); +} + +Note the return type: Map. +Any 5.x Partitioner implementation that imports org.springframework.batch.item.ExecutionContext +fails to compile against 6.0 with "cannot find symbol" on that single import line -- the fix is a +one-line import change, but the error message does not say that; it just says the type does not +exist, which sends most people straight to a search engine instead of to the correct one-line fix. + +--- Two more classes that moved, found while writing this module's beans --- +org.springframework.batch.core.job.parameters.JobParametersBuilder (was org.springframework.batch.core.JobParametersBuilder) +org.springframework.batch.core.repository.explore.JobExplorer (was org.springframework.batch.core.explore.JobExplorer) +org.springframework.batch.core.scope.context.StepSynchronizationManager (NOT org.springframework.batch.core.step -- BatchConfig's + first draft guessed that package for reading the current partition's step name inside an + ItemWriter and failed to compile; see docs/04-the-writer-and-the-beanmapped-trap.md) diff --git a/spring-batch-partitioning/docs/output/04-enforceuniquemethods-error.txt b/spring-batch-partitioning/docs/output/04-enforceuniquemethods-error.txt new file mode 100644 index 0000000..91807d1 --- /dev/null +++ b/spring-batch-partitioning/docs/output/04-enforceuniquemethods-error.txt @@ -0,0 +1,23 @@ +# Overloading a @Bean method across mutually-exclusive @Profile beans: rejected at startup + +First draft of the two TaskExecutor beans in BatchConfig used the same method name, +`partitionTaskExecutor`, distinguished only by @Profile("!reject") / @Profile("reject"). +Spring Framework 7's @Configuration.enforceUniqueMethods (on by default) does not know the two +profiles are mutually exclusive at class-parsing time -- it only sees one method name declared +twice -- and refuses to start: + +$ java -jar target/spring-batch-partitioning-1.0.0.jar --partition.shards-dir=./data/shards-small +... +2026-09-14T09:06:49.013Z WARN 2629 --- [ main] ConfigServletWebServerApplicationContext : Exception encountered during context initialization - cancelling refresh attempt: org.springframework.beans.factory.parsing.BeanDefinitionParsingException: Configuration problem: @Configuration class 'BatchConfig' contains overloaded @Bean methods with name 'partitionTaskExecutor'. Use unique method names for separate bean definitions (with individual conditions etc) or switch '@Configuration.enforceUniqueMethods' to 'false'. +Offending resource: class path resource [com/ankurm/batchpartition/config/BatchConfig.class] + +org.springframework.beans.factory.parsing.BeanDefinitionParsingException: Configuration problem: @Configuration class 'BatchConfig' contains overloaded @Bean methods with name 'partitionTaskExecutor'. Use unique method names for separate bean definitions (with individual conditions etc) or switch '@Configuration.enforceUniqueMethods' to 'false'. + at org.springframework.beans.factory.parsing.FailFastProblemReporter.error(FailFastProblemReporter.java:71) ~[spring-beans-7.0.9.jar!/:7.0.9] + at org.springframework.context.annotation.ConfigurationClass.validate(ConfigurationClass.java:265) ~[spring-context-7.0.9.jar!/:7.0.9] + at org.springframework.context.annotation.ConfigurationClassParser.validate(ConfigurationClassParser.java:230) ~[spring-context-7.0.9.jar!/:7.0.9] + +The message is accurate and the fix it suggests (unique method names) is the right one -- this +module's fixed version names the two beans partitionTaskExecutor() and +partitionTaskExecutorRejecting(int). The point worth recording: this is NOT a Spring Batch 6 +change, it is a Spring Framework 7 @Configuration default that bites a pattern (profile-gated +@Bean overloads) that plenty of Spring Batch 5.x tutorials use freely. diff --git a/spring-batch-partitioning/docs/output/05-happy-path-4-partitions.txt b/spring-batch-partitioning/docs/output/05-happy-path-4-partitions.txt new file mode 100644 index 0000000..ca47417 --- /dev/null +++ b/spring-batch-partitioning/docs/output/05-happy-path-4-partitions.txt @@ -0,0 +1,65 @@ +# Happy path: 4 shard files, gridSize 4, pool size 4 -- 20,000 rows + +$ java -jar target/spring-batch-partitioning-1.0.0.jar \ + --partition.shards-dir=./data/shards-small --partition.grid-size=4 \ + --partition.pool-core-size=4 --partition.pool-max-size=4 + +2026-09-14T09:29:56.758Z INFO 4937 --- [ main] c.a.b.PartitioningDemoApplication : Started PartitioningDemoApplication in 3.563 seconds (process running for 4.114) +2026-09-14T09:29:56.876Z INFO 4937 --- [ main] o.s.batch.core.step.AbstractStep : Executing step: [ordersManagerStep] +2026-09-14T09:29:56.892Z INFO 4937 --- [der-partition-2] o.s.batch.core.step.AbstractStep : Executing step: [ordersWorkerStep:partition0] +2026-09-14T09:29:56.890Z INFO 4937 --- [der-partition-1] o.s.batch.core.step.AbstractStep : Executing step: [ordersWorkerStep:partition1] +2026-09-14T09:29:56.900Z INFO 4937 --- [der-partition-4] o.s.batch.core.step.AbstractStep : Executing step: [ordersWorkerStep:partition2] +2026-09-14T09:29:56.901Z INFO 4937 --- [der-partition-3] o.s.batch.core.step.AbstractStep : Executing step: [ordersWorkerStep:partition3] +2026-09-14T09:29:57.962Z INFO 4937 --- [der-partition-2] o.s.batch.core.step.AbstractStep : Step: [ordersWorkerStep:partition0] executed in 1s70ms +2026-09-14T09:29:57.963Z INFO 4937 --- [der-partition-4] o.s.batch.core.step.AbstractStep : Step: [ordersWorkerStep:partition2] executed in 1s64ms +2026-09-14T09:29:57.990Z INFO 4937 --- [der-partition-1] o.s.batch.core.step.AbstractStep : Step: [ordersWorkerStep:partition1] executed in 1s99ms +2026-09-14T09:29:57.998Z INFO 4937 --- [der-partition-3] o.s.batch.core.step.AbstractStep : Step: [ordersWorkerStep:partition3] executed in 1s98ms +2026-09-14T09:29:58.005Z INFO 4937 --- [ main] o.s.batch.core.step.AbstractStep : Step: [ordersManagerStep] executed in 1s130ms +2026-09-14T09:29:58.011Z INFO 4937 --- [ main] o.s.batch.core.step.AbstractStep : Executing step: [reportStep] +REPORT: 20000 orders scored, 1561 flagged high-risk +JOB FINISHED: id=1 status=COMPLETED exitCode=COMPLETED + +All four partitions start within a few milliseconds of each other, on four distinct named +threads (order-partition-1..4) -- this is what "partitioned" actually looks like at the OS level, +not just in configuration. 1,561 of 20,000 orders (7.8%) came back flagged high-risk, matching the +expected rate from the threshold math in docs/06-why-cpu-bound-not-io-bound.md. + +--- GET /batch/partitions/1 (the diagnostic endpoint -- delete before shipping) --- +[ + { + "PARTITION_NAME": "ordersWorkerStep:partition0", + "THREAD_NAME": "order-partition-2", + "READ_COUNT": 5000, + "STARTED_AT": "2026-09-14T09:29:56.892Z", + "FINISHED_AT": "2026-09-14T09:29:57.968Z", + "DURATION_MS": 1075, + "EXIT_CODE": "COMPLETED" + }, + { + "PARTITION_NAME": "ordersWorkerStep:partition1", + "THREAD_NAME": "order-partition-1", + "READ_COUNT": 5000, + "STARTED_AT": "2026-09-14T09:29:56.895Z", + "FINISHED_AT": "2026-09-14T09:29:57.999Z", + "DURATION_MS": 1103, + "EXIT_CODE": "COMPLETED" + }, + { + "PARTITION_NAME": "ordersWorkerStep:partition2", + "THREAD_NAME": "order-partition-4", + "READ_COUNT": 5000, + "STARTED_AT": "2026-09-14T09:29:56.900Z", + "FINISHED_AT": "2026-09-14T09:29:57.964Z", + "DURATION_MS": 1064, + "EXIT_CODE": "COMPLETED" + }, + { + "PARTITION_NAME": "ordersWorkerStep:partition3", + "THREAD_NAME": "order-partition-3", + "READ_COUNT": 5000, + "STARTED_AT": "2026-09-14T09:29:56.901Z", + "FINISHED_AT": "2026-09-14T09:29:58.002Z", + "DURATION_MS": 1101, + "EXIT_CODE": "COMPLETED" + } +] diff --git a/spring-batch-partitioning/docs/output/06-rejected-partitions-stuck.txt b/spring-batch-partitioning/docs/output/06-rejected-partitions-stuck.txt new file mode 100644 index 0000000..cf8dc7a --- /dev/null +++ b/spring-batch-partitioning/docs/output/06-rejected-partitions-stuck.txt @@ -0,0 +1,33 @@ +# Undersized thread pool + AbortPolicy: 3 of 4 partitions rejected, and stuck forever + +$ java -jar target/spring-batch-partitioning-1.0.0.jar --spring.profiles.active=reject \ + --partition.shards-dir=./data/shards-small --partition.grid-size=4 --partition.reject.pool-size=1 + +2026-09-14T09:11:18.181Z INFO 3074 --- [ main] c.a.b.PartitioningDemoApplication : The following 1 profile is active: "reject" +2026-09-14T09:11:21.361Z INFO 3074 --- [ main] o.s.batch.core.step.AbstractStep : Executing step: [ordersManagerStep] +2026-09-14T09:11:21.375Z INFO 3074 --- [der-partition-1] o.s.batch.core.step.AbstractStep : Executing step: [ordersWorkerStep:partition1] +2026-09-14T09:11:21.946Z INFO 3074 --- [der-partition-1] o.s.batch.core.step.AbstractStep : Step: [ordersWorkerStep:partition1] executed in 570ms +org.springframework.batch.core.job.JobExecutionException: Partition handler returned an unsuccessful step +2026-09-14T09:11:21.961Z INFO 3074 --- [ main] o.s.batch.core.step.AbstractStep : Step: [ordersManagerStep] executed in 601ms +JOB FINISHED: id=1 status=FAILED exitCode=FAILED + +Only ordersWorkerStep:partition1 ever logs "Executing step" -- the other three were rejected by +the ThreadPoolTaskExecutor (corePoolSize=1, queueCapacity=0, AbortPolicy) before they could even +start, and TaskExecutorPartitionHandler swallows that TaskRejectedException into the rejected +StepExecution's failure list rather than printing it -- nothing named "TaskRejectedException" or +"Rejected" ever appears in this log. The job fails with the generic message above. + +--- Querying BATCH_STEP_EXECUTION directly (org.h2.tools.Shell) shows what the log does not --- +$ java -cp h2-2.4.240.jar org.h2.tools.Shell -url jdbc:h2:file:./data/rejecttest -user sa -password "" \ + -sql "SELECT STEP_EXECUTION_ID, STEP_NAME, STATUS, EXIT_CODE FROM BATCH_STEP_EXECUTION ORDER BY STEP_EXECUTION_ID;" + +STEP_EXECUTION_ID | STEP_NAME | STATUS | EXIT_CODE +1 | ordersManagerStep | FAILED | FAILED +2 | ordersWorkerStep:partition3 | STARTING | EXECUTING +3 | ordersWorkerStep:partition2 | STARTING | EXECUTING +4 | ordersWorkerStep:partition1 | COMPLETED | COMPLETED +5 | ordersWorkerStep:partition0 | STARTING | EXECUTING + +The manager step and the job both reach FAILED. The three rejected worker StepExecutions do not +-- they are parked at STARTING/EXECUTING permanently. Nothing in this job's lifecycle ever +transitions them again on its own. diff --git a/spring-batch-partitioning/docs/output/07-restart-throws-alreadyrunning.txt b/spring-batch-partitioning/docs/output/07-restart-throws-alreadyrunning.txt new file mode 100644 index 0000000..82b68d5 --- /dev/null +++ b/spring-batch-partitioning/docs/output/07-restart-throws-alreadyrunning.txt @@ -0,0 +1,18 @@ +# Restarting the stuck job: JobExecutionAlreadyRunningException, forever + +$ java -jar target/spring-batch-partitioning-1.0.0.jar \ + --partition.shards-dir=./data/shards-small --partition.grid-size=4 --partition.pool-core-size=4 --partition.pool-max-size=4 +(same shardsDir = same identifying job parameter = same JobInstance = restart target) + +Caused by: org.springframework.batch.core.launch.JobExecutionAlreadyRunningException: A job execution for this job is already running: JobExecution: id=1, version=3, startTime=2026-09-14T09:11:21.331474360, endTime=2026-09-14T09:11:21.965090481, lastUpdated=2026-09-14T09:11:21.966521177, status=FAILED, exitStatus=exitCode=FAILED;exitDescription=org.springframework.batch.core.job.JobExecutionException: Partition handler returned an unsuccessful step + at org.springframework.batch.core.partition.PartitionStep.doExecute(PartitionStep.java:134) + at org.springframework.batch.core.step.AbstractStep.execute(AbstractStep.java:251) + + +The JobExecution row plainly says status=FAILED right there in the exception's own message, and +SimpleJobOperator still refuses to start a new attempt against it, because its check for "is this +JobInstance already running" is not "is the JobExecution FAILED" -- it is closer to "does this +instance have any StepExecution that is not in a terminal status", and the three orphaned +STARTING/EXECUTING worker steps from docs/06-rejected-partitions-stuck.txt are exactly that. Every +subsequent `java -jar ... ` against the same shards-dir throws this same exception. The job is +not failed. It is stuck. diff --git a/spring-batch-partitioning/docs/output/08-recover-then-restart.txt b/spring-batch-partitioning/docs/output/08-recover-then-restart.txt new file mode 100644 index 0000000..0779399 --- /dev/null +++ b/spring-batch-partitioning/docs/output/08-recover-then-restart.txt @@ -0,0 +1,46 @@ +# Spring Batch 6.0's fix: JobOperator#recover, then a normal restart + +$ java -jar target/spring-batch-partitioning-1.0.0.jar --spring.profiles.active=recover \ + --recover.job-execution-id=1 --partition.shards-dir=./data/shards-small \ + --partition.grid-size=4 --partition.pool-core-size=4 --partition.pool-max-size=4 + +RECOVER: before -> status=FAILED +RECOVER: step=ordersManagerStep status=FAILED +RECOVER: step=ordersWorkerStep:partition3 status=STARTING +RECOVER: step=ordersWorkerStep:partition2 status=STARTING +RECOVER: step=ordersWorkerStep:partition1 status=COMPLETED +RECOVER: step=ordersWorkerStep:partition0 status=STARTING +2026-09-14T09:13:31.156Z INFO 3377 --- [ main] o.s.b.c.l.s.TaskExecutorJobOperator : Recovering job execution: JobExecution: id=1, version=3, startTime=2026-09-14T09:11:21.331474360, endTime=2026-09-14T09:11:21.965090481, lastUpdated=2026-09-14T09:11:21.966521177, status=FAILED, exitStatus=exitCode=FAILED;exitDescription=org.springframework.batch.core.job.JobExecutionException: Partition handler returned an unsuccessful step +RECOVER: after -> status=FAILED +RECOVER: step=ordersManagerStep status=FAILED +RECOVER: step=ordersWorkerStep:partition3 status=FAILED +RECOVER: step=ordersWorkerStep:partition2 status=FAILED +RECOVER: step=ordersWorkerStep:partition1 status=COMPLETED +RECOVER: step=ordersWorkerStep:partition0 status=FAILED +JOB FINISHED: id=33 status=COMPLETED exitCode=COMPLETED + +recover() walked the stuck JobExecution's StepExecutions and force-closed the three still at +STARTING to FAILED -- nothing else changed. RecoveryRunner runs at @Order(0); OrderIngestRunner +then runs its normal start() immediately after, in the same JVM, against the same shardsDir, and +this time it succeeds: a brand-new JobExecution (id=33) completes. + +--- BATCH_STEP_EXECUTION after recovery + restart: which partitions actually reran --- +$ java -cp h2-2.4.240.jar org.h2.tools.Shell -url jdbc:h2:file:./data/rejecttest -user sa -password "" \ + -sql "SELECT JOB_EXECUTION_ID, STEP_EXECUTION_ID, STEP_NAME, STATUS, READ_COUNT FROM BATCH_STEP_EXECUTION WHERE JOB_EXECUTION_ID IN (1,33) ORDER BY JOB_EXECUTION_ID, STEP_EXECUTION_ID;" + +JOB_EXECUTION_ID | STEP_EXECUTION_ID | STEP_NAME | STATUS | READ_COUNT +1 | 1 | ordersManagerStep | FAILED | 5000 +1 | 2 | ordersWorkerStep:partition3 | FAILED | 0 +1 | 3 | ordersWorkerStep:partition2 | FAILED | 0 +1 | 4 | ordersWorkerStep:partition1 | COMPLETED | 5000 +1 | 5 | ordersWorkerStep:partition0 | FAILED | 0 +33 | 33 | ordersManagerStep | COMPLETED | 15000 +33 | 34 | ordersWorkerStep:partition3 | COMPLETED | 5000 +33 | 35 | ordersWorkerStep:partition2 | COMPLETED | 5000 +33 | 36 | ordersWorkerStep:partition0 | COMPLETED | 5000 +33 | 37 | reportStep | COMPLETED | 0 + +JobExecution 33 has exactly three new worker StepExecutions -- partition3, partition2, partition0, +the ones recover() marked FAILED. There is no new StepExecution for partition1: it stayed +COMPLETED from JobExecution 1 and was correctly skipped. Restart-only-the-failed-partition is not +a promise in the reference docs here -- it is what this table shows actually happened. diff --git a/spring-batch-partitioning/docs/output/09-full-scale-throughput.txt b/spring-batch-partitioning/docs/output/09-full-scale-throughput.txt new file mode 100644 index 0000000..d0cdc94 --- /dev/null +++ b/spring-batch-partitioning/docs/output/09-full-scale-throughput.txt @@ -0,0 +1,40 @@ +# 10,000,000 rows: manager-step wall time by partition count (2 vCPUs) + +Same job, same code path, same data (deterministic seed 100) -- only partition.grid-size and +the shard directory (which controls how many resources MultiResourcePartitioner sees) change. +risk.iterations=150 (the default) for all four runs. Each number is the 'Step: [ordersManagerStep] +executed in ...' line from that run's own log -- wall time for the whole partitioned step, +including every worker partition and the manager step's own bookkeeping. + +grid-size manager-step rows/sec speedup-vs-1 +1 70.485s 141,874 1.00x +2 49.647s 201,422 1.42x +4 53.102s 188,317 1.33x +8 55.521s 180,112 1.27x + +Best result at grid-size 2 -- matching this sandbox's 2 vCPUs exactly. Beyond that, wall time gets +WORSE with every doubling: grid-size 8 is slower than grid-size 4, which is slower than grid-size +2. More partitions past the physical core count does not sit still, it actively costs time -- +context-switch and scheduling overhead with no additional CPU to absorb it. The speedup at +grid-size 2 (1.42x) is also well short of the 2x a naive "twice the cores" mental model predicts; +docs/06-why-cpu-bound-not-io-bound.md and docs/10-scaling-sensitivity-to-data-size.md discuss the +two candidate reasons this article checked (H2's single-writer MVStore, and fixed per-partition +startup cost) and what evidence separates them. + +--- The same sweep at a smaller scale (300,000 rows) tells a different story --- +grid-size 1: 5.127s grid-size 2: 4.677s (1.10x) grid-size 4: 5.158s (0.99x) grid-size 8: 6.812s (0.75x) + +At 300K rows, grid-size 8 is not just worse than grid-size 2 -- it is worse than NOT partitioning +at all. The fixed cost of standing up a partition (opening the shard file, acquiring a JDBC +connection, thread handoff) is the same few milliseconds whether a job processes 10,000,000 rows +or 300,000; at the smaller scale there is less real work to amortize it against, so +over-partitioning is a strictly worse mistake on a smaller job than on a larger one. Whether +partitioning helps at all is a function of BOTH core count and data volume, not core count alone. + +--- A CPU-heavier variant (risk.iterations=5000, 300,000 rows) narrows the gap towards 2x --- +grid-size 1: 9.184s grid-size 2: 7.277s (1.26x) + +Raising the per-item CPU cost pushed the grid-size-2 speedup from 1.10x to 1.26x at the same data +volume -- evidence, not proof, that at least part of the shortfall from a clean 2x is the shared +H2 writer, not thread overhead alone: more CPU-bound work per item dilutes the fixed write-lock +cost relative to total time, and the measured speedup moved in exactly that direction. diff --git a/spring-batch-partitioning/pom.xml b/spring-batch-partitioning/pom.xml new file mode 100644 index 0000000..70b6ae6 --- /dev/null +++ b/spring-batch-partitioning/pom.xml @@ -0,0 +1,76 @@ + + + 4.0.0 + + + org.springframework.boot + spring-boot-starter-parent + 4.1.1 + + + + com.ankurm + spring-batch-partitioning + 1.0.0 + spring-batch-partitioning + Spring Batch partitioning and parallel steps: scaling a 10-million-row job + + + 25 + + + + + + org.springframework.boot + spring-boot-starter-batch-jdbc + + + + org.springframework.boot + spring-boot-starter-jdbc + + + + + org.springframework.boot + spring-boot-starter-web + + + + + com.h2database + h2 + runtime + + + + org.springframework.boot + spring-boot-starter-test + test + + + + org.springframework.batch + spring-batch-test + test + + + + + + + org.springframework.boot + spring-boot-maven-plugin + + + + diff --git a/spring-batch-partitioning/scripts/generate-shards.py b/spring-batch-partitioning/scripts/generate-shards.py new file mode 100755 index 0000000..43d35c5 --- /dev/null +++ b/spring-batch-partitioning/scripts/generate-shards.py @@ -0,0 +1,61 @@ +#!/usr/bin/env python3 +"""Generates the sharded order CSVs this module partitions over. + +Usage: + generate-shards.py [--corrupt-shard N] [--seed S] + +Each shard is orders-shard-NN.csv with header "orderId,customerId,amountCents,region". +Row counts are split as evenly as possible across shards (the last shard absorbs the remainder). +--corrupt-shard N replaces one row near the middle of shard N (0-indexed) with a non-numeric +amountCents field, so FlatFileItemReader's FieldSetMapper throws NumberFormatException wrapped in +FlatFileParseException when that shard is read -- a real, reproducible parse failure, not a +simulated one. +""" +import argparse +import os +import random +import sys + +REGIONS = ["NORTH", "SOUTH", "EAST", "WEST", "CENTRAL"] + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("output_dir") + ap.add_argument("total_rows", type=int) + ap.add_argument("num_shards", type=int) + ap.add_argument("--corrupt-shard", type=int, default=-1) + ap.add_argument("--seed", type=int, default=42) + args = ap.parse_args() + + os.makedirs(args.output_dir, exist_ok=True) + rnd = random.Random(args.seed) + + base = args.total_rows // args.num_shards + counts = [base] * args.num_shards + counts[-1] += args.total_rows - base * args.num_shards + + order_id = 1 + for shard_idx, count in enumerate(counts): + path = os.path.join(args.output_dir, f"orders-shard-{shard_idx:02d}.csv") + corrupt_at = -1 + if shard_idx == args.corrupt_shard: + corrupt_at = count // 2 + with open(path, "w", newline="") as f: + f.write("orderId,customerId,amountCents,region\n") + for i in range(count): + customer_id = rnd.randint(1, 2_000_000) + amount_cents = rnd.randint(500, 9_999_999) + region = REGIONS[rnd.randint(0, len(REGIONS) - 1)] + if i == corrupt_at: + f.write(f"{order_id},{customer_id},NOT_A_NUMBER,{region}\n") + else: + f.write(f"{order_id},{customer_id},{amount_cents},{region}\n") + order_id += 1 + print(f"wrote {path}: {count} rows" + (" (1 corrupt row)" if corrupt_at >= 0 else "")) + + print(f"total: {order_id - 1} rows across {args.num_shards} shards in {args.output_dir}") + + +if __name__ == "__main__": + main() diff --git a/spring-batch-partitioning/scripts/run.sh b/spring-batch-partitioning/scripts/run.sh new file mode 100755 index 0000000..4835246 --- /dev/null +++ b/spring-batch-partitioning/scripts/run.sh @@ -0,0 +1,28 @@ +#!/usr/bin/env bash +# Runs the demo once, in the foreground, streaming to both stdout and a log file. +# +# scripts/run.sh [grid-size] [pool-size] [extra java -D or --args...] +# +# Never `pkill -f 'spring-boot'` here -- the pattern matches this script's own invocation line +# under some shells and can kill the wrong process. Kill by main class instead. +set -eu +DB_NAME="${1:?usage: run.sh [grid-size] [pool-size] [extra args...]}" +SHARDS_DIR="${2:?usage: run.sh [grid-size] [pool-size] [extra args...]}" +GRID_SIZE="${3:-4}" +POOL_SIZE="${4:-4}" +shift $(( $# >= 4 ? 4 : $# )) + +for p in $(ps -eo pid,cmd | grep '[P]artitioningDemoApplication' | awk '{print $1}'); do + kill -9 "$p" 2>/dev/null || true +done + +JAR="$(dirname "$0")/../target/spring-batch-partitioning-1.0.0.jar" +LOG="/tmp/run-${DB_NAME}.log" + +"${JAVA_HOME:?set JAVA_HOME}/bin/java" -jar "$JAR" \ + --spring.datasource.url="jdbc:h2:file:./data/${DB_NAME};AUTO_SERVER=TRUE" \ + --partition.shards-dir="$SHARDS_DIR" \ + --partition.grid-size="$GRID_SIZE" \ + --partition.pool-core-size="$POOL_SIZE" \ + --partition.pool-max-size="$POOL_SIZE" \ + "$@" 2>&1 | tee "$LOG" diff --git a/spring-batch-partitioning/src/main/java/com/ankurm/batchpartition/PartitioningDemoApplication.java b/spring-batch-partitioning/src/main/java/com/ankurm/batchpartition/PartitioningDemoApplication.java new file mode 100644 index 0000000..fd114f6 --- /dev/null +++ b/spring-batch-partitioning/src/main/java/com/ankurm/batchpartition/PartitioningDemoApplication.java @@ -0,0 +1,11 @@ +package com.ankurm.batchpartition; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; + +@SpringBootApplication +public class PartitioningDemoApplication { + public static void main(String[] args) { + SpringApplication.run(PartitioningDemoApplication.class, args); + } +} diff --git a/spring-batch-partitioning/src/main/java/com/ankurm/batchpartition/config/BatchConfig.java b/spring-batch-partitioning/src/main/java/com/ankurm/batchpartition/config/BatchConfig.java new file mode 100644 index 0000000..0d05d7c --- /dev/null +++ b/spring-batch-partitioning/src/main/java/com/ankurm/batchpartition/config/BatchConfig.java @@ -0,0 +1,257 @@ +package com.ankurm.batchpartition.config; + +import com.ankurm.batchpartition.domain.Order; +import com.ankurm.batchpartition.domain.RiskScoredOrder; +import com.ankurm.batchpartition.partition.PartitionStatsListener; +import com.ankurm.batchpartition.processing.RiskScoringProcessor; +import org.springframework.batch.core.configuration.annotation.StepScope; +import org.springframework.batch.core.job.Job; +import org.springframework.batch.core.job.builder.JobBuilder; +import org.springframework.batch.core.partition.Partitioner; +import org.springframework.batch.core.partition.support.MultiResourcePartitioner; +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.FlatFileItemReader; +import org.springframework.batch.infrastructure.item.file.builder.FlatFileItemReaderBuilder; +import org.springframework.batch.infrastructure.repeat.RepeatStatus; +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.core.io.support.PathMatchingResourcePatternResolver; +import org.springframework.core.task.TaskExecutor; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; +import org.springframework.transaction.PlatformTransactionManager; + +import java.io.IOException; +import java.util.concurrent.ThreadPoolExecutor; + +/** + * One job, {@code orderRiskJob}: partition a directory of pre-sharded order CSVs across worker + * threads, score every order for risk, write the result to {@code ORDER_RISK_SUMMARY}. + * + *

The manager/worker split follows {@code docs/02-anatomy-of-a-partitioned-step.md}. The + * single most load-bearing design choice in this module: there is no separately-coded + * "single-threaded baseline". Running with {@code --partition.grid-size=1} against one shard + * file drives the exact same {@link #partitioner}/{@link #workerStep}/{@link #managerStep} code + * path as running with grid size 8 against eight shards — so the numbers in the article + * differ by exactly one variable (thread count), not by two (thread count AND a different code + * path). See {@code docs/01-the-problem-and-mental-model.md}. + */ +@Configuration +public class BatchConfig { + + @Value("${partition.shards-dir}") + private String shardsDir; + + @Value("${partition.grid-size:4}") + private int gridSize; + + @Value("${partition.pool-core-size:4}") + private int poolCoreSize; + + @Value("${partition.pool-max-size:4}") + private int poolMaxSize; + + @Value("${partition.pool-queue-capacity:0}") + private int poolQueueCapacity; + + @Value("${risk.iterations:150}") + private int riskIterations; + + @Value("${risk.high-risk-amount-cents:5000000}") + private long highRiskAmountCents; + + // ---- partitioning: one ExecutionContext per shard file --------------------------------- + + /** + * {@link MultiResourcePartitioner#partition(int)} ignores the {@code gridSize} argument it is + * handed — verified by decompiling {@code spring-batch-core-6.0.5.jar}: the method body + * loops over the configured {@code resources} array and never reads its {@code int} parameter + * at all. The number of partitions this job runs is the number of shard files in + * {@code partition.shards-dir}, full stop. {@code docs/03-what-gridsize-actually-controls.md} + * has the decompiled bytecode and the run that proves it (three shard files, gridSize 10, + * three partitions). + */ + @Bean + public Partitioner partitioner() throws IOException { + var resolver = new PathMatchingResourcePatternResolver(); + Resource[] resources = resolver.getResources("file:" + shardsDir + "/*.csv"); + java.util.Arrays.sort(resources, java.util.Comparator.comparing(r -> { + try { + return r.getFilename(); + } catch (Exception e) { + return ""; + } + })); + var partitioner = new MultiResourcePartitioner(); + partitioner.setResources(resources); + partitioner.setKeyName("fileName"); + return partitioner; + } + + // ---- worker step: reads ONE shard file, scores it, writes it --------------------------- + + @Bean + @StepScope + public FlatFileItemReader shardReader(@Value("#{stepExecutionContext['fileName']}") Resource shardFile) { + return new FlatFileItemReaderBuilder() + .name("shardReader") + .resource(shardFile) + .linesToSkip(1) + .delimited().delimiter(",").names("orderId", "customerId", "amountCents", "region") + .fieldSetMapper(fs -> new Order( + fs.readLong("orderId"), + fs.readLong("customerId"), + fs.readLong("amountCents"), + fs.readString("region"))) + .build(); + } + + @Bean + public ItemProcessor riskProcessor() { + return new RiskScoringProcessor(riskIterations, highRiskAmountCents); + } + + /** + * Not {@code beanMapped()}, same reasoning {@code spring-batch/} (the earlier module) already + * found: {@code BeanPropertySqlParameterSource} looks for JavaBean getters, and a record's + * accessors are {@code orderId()} not {@code getOrderId()}. + * + *

{@code partitionName} is late-bound from {@code #{stepExecution.stepName}}. The first + * draft of this bean reached for {@code StepSynchronizationManager} to read the current step + * name instead — {@code javap} on the real 6.0.5 jar showed that class does not exist at + * {@code org.springframework.batch.core.step.StepSynchronizationManager}; it is at + * {@code org.springframework.batch.core.scope.context.StepSynchronizationManager}, one of + * several classes this article's research moved out of the {@code core.step} package. Late + * binding sidesteps the question entirely and is the idiomatic way to reach step identity from + * a step-scoped bean anyway. + */ + @Bean + @StepScope + public ItemWriter riskWriter(JdbcTemplate jdbcTemplate, + @Value("#{stepExecution.stepName}") String partitionName) { + return new JdbcBatchItemWriterBuilder() + .dataSource(jdbcTemplate.getDataSource()) + .sql("MERGE INTO ORDER_RISK_SUMMARY (order_id, customer_id, amount_cents, region, risk_score, " + + "high_risk, partition_name) KEY(order_id) VALUES (?, ?, ?, ?, ?, ?, ?)") + .itemPreparedStatementSetter((item, ps) -> { + ps.setLong(1, item.order().orderId()); + ps.setLong(2, item.order().customerId()); + ps.setLong(3, item.order().amountCents()); + ps.setString(4, item.order().region()); + ps.setInt(5, item.riskScore()); + ps.setBoolean(6, item.highRisk()); + ps.setString(7, partitionName); + }) + .assertUpdates(true) + .build(); + } + + @Bean + public PartitionStatsListener partitionStatsListener(JdbcTemplate jdbcTemplate) { + return new PartitionStatsListener(jdbcTemplate); + } + + @Bean + public Step workerStep(JobRepository jobRepository, PlatformTransactionManager transactionManager, + FlatFileItemReader shardReader, ItemProcessor riskProcessor, + ItemWriter riskWriter, PartitionStatsListener partitionStatsListener) { + return new StepBuilder("ordersWorkerStep", jobRepository) + .chunk(1000) + .transactionManager(transactionManager) + .reader(shardReader) + .processor(riskProcessor) + .writer(riskWriter) + .listener(partitionStatsListener) + .build(); + } + + // ---- the two task executors: the working one, and the one that demonstrates rejection -- + + @Bean + @Profile("!reject") + public TaskExecutor partitionTaskExecutor() { + var executor = new ThreadPoolTaskExecutor(); + executor.setCorePoolSize(poolCoreSize); + executor.setMaxPoolSize(poolMaxSize); + executor.setQueueCapacity(poolQueueCapacity); + executor.setThreadNamePrefix("order-partition-"); + executor.initialize(); + return executor; + } + + /** + * {@code docs/07-the-rejectedexecutionexception.md}: a pool sized smaller than the number of + * shard files, a zero-capacity queue, and {@link ThreadPoolExecutor.AbortPolicy} — the + * combination {@link org.springframework.batch.core.partition.support.TaskExecutorPartitionHandler} + * needs to actually throw {@code TaskRejectedException} instead of silently serialising the + * "extra" partitions onto the caller thread (which is what + * {@link ThreadPoolExecutor.CallerRunsPolicy}, Spring's default rejection handler, does). + * + *

This bean was originally an overload of {@link #partitionTaskExecutor()} distinguished + * only by {@code @Profile}. Spring Framework 7's {@code @Configuration.enforceUniqueMethods} + * (on by default) rejects that at startup with {@code BeanDefinitionParsingException: + * contains overloaded @Bean methods} — it does not know the two profiles are mutually + * exclusive at parse time, only that the method name collides. The fix is a distinct method + * name, not a workaround; see {@code docs/07-the-rejectedexecutionexception.md} for the exact + * exception text this produced. + */ + @Bean + @Profile("reject") + public TaskExecutor partitionTaskExecutorRejecting(@Value("${partition.reject.pool-size:1}") int smallPoolSize) { + var executor = new ThreadPoolTaskExecutor(); + executor.setCorePoolSize(smallPoolSize); + executor.setMaxPoolSize(smallPoolSize); + executor.setQueueCapacity(0); + executor.setThreadNamePrefix("order-partition-"); + executor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy()); + executor.initialize(); + return executor; + } + + // ---- manager step + job ----------------------------------------------------------------- + + @Bean + public Step managerStep(JobRepository jobRepository, Partitioner partitioner, Step workerStep, + TaskExecutor partitionTaskExecutor) { + return new StepBuilder("ordersManagerStep", jobRepository) + .partitioner("ordersWorkerStep", partitioner) + .step(workerStep) + .gridSize(gridSize) + .taskExecutor(partitionTaskExecutor) + .build(); + } + + @Bean + public Tasklet reportTasklet(JdbcTemplate jdbcTemplate) { + return (contribution, chunkContext) -> { + int total = jdbcTemplate.queryForObject("SELECT COUNT(*) FROM ORDER_RISK_SUMMARY", Integer.class); + int highRisk = jdbcTemplate.queryForObject( + "SELECT COUNT(*) FROM ORDER_RISK_SUMMARY WHERE high_risk = TRUE", Integer.class); + System.out.println("REPORT: " + total + " orders scored, " + highRisk + " flagged high-risk"); + return RepeatStatus.FINISHED; + }; + } + + @Bean + public Step reportStep(JobRepository jobRepository, PlatformTransactionManager transactionManager, + Tasklet reportTasklet) { + return new StepBuilder("reportStep", jobRepository) + .tasklet(reportTasklet, transactionManager) + .allowStartIfComplete(true) + .build(); + } + + @Bean + public Job orderRiskJob(JobRepository jobRepository, Step managerStep, Step reportStep) { + return new JobBuilder("orderRiskJob", jobRepository).start(managerStep).next(reportStep).build(); + } +} diff --git a/spring-batch-partitioning/src/main/java/com/ankurm/batchpartition/domain/Order.java b/spring-batch-partitioning/src/main/java/com/ankurm/batchpartition/domain/Order.java new file mode 100644 index 0000000..e22e58c --- /dev/null +++ b/spring-batch-partitioning/src/main/java/com/ankurm/batchpartition/domain/Order.java @@ -0,0 +1,9 @@ +package com.ankurm.batchpartition.domain; + +/** + * One row of the input CSV. A record, deliberately — see + * {@code docs/04-the-writer-and-the-beanmapped-trap.md} for why {@link #writer} in + * {@link com.ankurm.batchpartition.config.BatchConfig} does not use {@code beanMapped()}. + */ +public record Order(long orderId, long customerId, long amountCents, String region) { +} diff --git a/spring-batch-partitioning/src/main/java/com/ankurm/batchpartition/domain/RiskScoredOrder.java b/spring-batch-partitioning/src/main/java/com/ankurm/batchpartition/domain/RiskScoredOrder.java new file mode 100644 index 0000000..4303267 --- /dev/null +++ b/spring-batch-partitioning/src/main/java/com/ankurm/batchpartition/domain/RiskScoredOrder.java @@ -0,0 +1,5 @@ +package com.ankurm.batchpartition.domain; + +/** {@link Order} plus what {@link com.ankurm.batchpartition.processing.RiskScoringProcessor} computed. */ +public record RiskScoredOrder(Order order, int riskScore, boolean highRisk) { +} diff --git a/spring-batch-partitioning/src/main/java/com/ankurm/batchpartition/partition/PartitionStatsListener.java b/spring-batch-partitioning/src/main/java/com/ankurm/batchpartition/partition/PartitionStatsListener.java new file mode 100644 index 0000000..d1d63f3 --- /dev/null +++ b/spring-batch-partitioning/src/main/java/com/ankurm/batchpartition/partition/PartitionStatsListener.java @@ -0,0 +1,53 @@ +package com.ankurm.batchpartition.partition; + +import org.springframework.batch.core.ExitStatus; +import org.springframework.batch.core.listener.StepExecutionListener; +import org.springframework.batch.core.step.StepExecution; +import org.springframework.jdbc.core.JdbcTemplate; + +import java.time.Duration; +import java.time.LocalDateTime; + +/** + * Records which thread ran which worker partition, and how long it took, into + * {@code PARTITION_STATS}. This is the hidden runtime state {@code /batch/partitions} in + * {@link com.ankurm.batchpartition.web.PartitionInsightController} exposes — without it, + * "did partitioning actually run four threads or one?" is a question you can only answer by + * trusting the configuration, not by looking at what happened. Delete this listener (and the + * controller) before shipping a real job; it exists here to make the mechanism visible, not + * because production batch jobs should query their own thread names. + */ +public class PartitionStatsListener implements StepExecutionListener { + + private final JdbcTemplate jdbcTemplate; + private final ThreadLocal startedAt = new ThreadLocal<>(); + + public PartitionStatsListener(JdbcTemplate jdbcTemplate) { + this.jdbcTemplate = jdbcTemplate; + } + + @Override + public void beforeStep(StepExecution stepExecution) { + startedAt.set(LocalDateTime.now()); + } + + @Override + public ExitStatus afterStep(StepExecution stepExecution) { + LocalDateTime start = startedAt.get(); + LocalDateTime end = LocalDateTime.now(); + long durationMs = start == null ? -1 : Duration.between(start, end).toMillis(); + jdbcTemplate.update( + "INSERT INTO PARTITION_STATS (job_execution_id, partition_name, thread_name, read_count, " + + "started_at, finished_at, duration_ms, exit_code) VALUES (?, ?, ?, ?, ?, ?, ?, ?)", + stepExecution.getJobExecutionId(), + stepExecution.getStepName(), + Thread.currentThread().getName(), + stepExecution.getReadCount(), + start, + end, + durationMs, + stepExecution.getExitStatus().getExitCode()); + startedAt.remove(); + return stepExecution.getExitStatus(); + } +} diff --git a/spring-batch-partitioning/src/main/java/com/ankurm/batchpartition/processing/RiskScoringProcessor.java b/spring-batch-partitioning/src/main/java/com/ankurm/batchpartition/processing/RiskScoringProcessor.java new file mode 100644 index 0000000..72fe623 --- /dev/null +++ b/spring-batch-partitioning/src/main/java/com/ankurm/batchpartition/processing/RiskScoringProcessor.java @@ -0,0 +1,37 @@ +package com.ankurm.batchpartition.processing; + +import com.ankurm.batchpartition.domain.Order; +import com.ankurm.batchpartition.domain.RiskScoredOrder; +import org.springframework.batch.infrastructure.item.ItemProcessor; + +/** + * Deliberately CPU-bound, not I/O-bound — this is what makes the partitioning numbers in + * this article mean something on a 2-core sandbox. A pure I/O-bound step (a network call per + * item) would show partitioning "working" even with a single core free, because the threads + * spend their time blocked, not computing. Here every item does {@code risk.iterations} real + * multiplications, so the speedup this module measures is bounded by actual core count, not by + * how many threads are merely alive. See {@code docs/06-why-cpu-bound-not-io-bound.md}. + */ +public class RiskScoringProcessor implements ItemProcessor { + + private final int iterations; + private final long highRiskAmountCents; + + public RiskScoringProcessor(int iterations, long highRiskAmountCents) { + this.iterations = iterations; + this.highRiskAmountCents = highRiskAmountCents; + } + + @Override + public RiskScoredOrder process(Order order) { + long acc = order.orderId() * 2654435761L + order.customerId(); + for (int i = 0; i < iterations; i++) { + acc = (acc ^ (acc >>> 13)) * 2246822519L; + acc = (acc ^ (acc >>> 15)) * 3266489917L; + acc = acc ^ (acc >>> 16); + } + int score = (int) Math.floorMod(acc, 1000); + boolean highRisk = order.amountCents() > highRiskAmountCents || score > 970; + return new RiskScoredOrder(order, score, highRisk); + } +} diff --git a/spring-batch-partitioning/src/main/java/com/ankurm/batchpartition/runner/OrderIngestRunner.java b/spring-batch-partitioning/src/main/java/com/ankurm/batchpartition/runner/OrderIngestRunner.java new file mode 100644 index 0000000..c5c9022 --- /dev/null +++ b/spring-batch-partitioning/src/main/java/com/ankurm/batchpartition/runner/OrderIngestRunner.java @@ -0,0 +1,47 @@ +package com.ankurm.batchpartition.runner; + +import org.springframework.batch.core.job.Job; +import org.springframework.batch.core.job.parameters.JobParametersBuilder; +import org.springframework.batch.core.launch.JobInstanceAlreadyCompleteException; +import org.springframework.batch.core.launch.JobOperator; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.ApplicationArguments; +import org.springframework.boot.ApplicationRunner; +import org.springframework.stereotype.Component; + +/** + * {@code partition.shards-dir} is the identifying job parameter: two runs against the same + * directory are the same {@code JobInstance}, so a run that failed partway through resumes + * (only the failed partitions re-execute) rather than starting over. Point + * {@code partition.shards-dir} at a different directory to force a fresh instance. See + * {@code docs/07-the-rejectedexecutionexception.md} and + * {@code docs/08-restart-reruns-only-the-failed-partition.md}. + */ +@Component +public class OrderIngestRunner implements ApplicationRunner { + + private final JobOperator jobOperator; + private final Job job; + + @Value("${partition.shards-dir}") + private String shardsDir; + + public OrderIngestRunner(JobOperator jobOperator, Job job) { + this.jobOperator = jobOperator; + this.job = job; + } + + @Override + public void run(ApplicationArguments args) throws Exception { + var params = new JobParametersBuilder() + .addString("shardsDir", shardsDir) // identifying: same dir = same JobInstance = restart target + .toJobParameters(); + try { + var execution = jobOperator.start(job, params); + System.out.println("JOB FINISHED: id=" + execution.getId() + " status=" + execution.getStatus() + + " exitCode=" + execution.getExitStatus().getExitCode()); + } catch (JobInstanceAlreadyCompleteException e) { + System.out.println("JOB ALREADY COMPLETE: " + e.getMessage()); + } + } +} diff --git a/spring-batch-partitioning/src/main/java/com/ankurm/batchpartition/runner/RecoveryRunner.java b/spring-batch-partitioning/src/main/java/com/ankurm/batchpartition/runner/RecoveryRunner.java new file mode 100644 index 0000000..99057d8 --- /dev/null +++ b/spring-batch-partitioning/src/main/java/com/ankurm/batchpartition/runner/RecoveryRunner.java @@ -0,0 +1,58 @@ +package com.ankurm.batchpartition.runner; + +import org.springframework.batch.core.job.JobExecution; +import org.springframework.batch.core.launch.JobOperator; +import org.springframework.batch.core.repository.explore.JobExplorer; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.ApplicationArguments; +import org.springframework.boot.ApplicationRunner; +import org.springframework.context.annotation.Profile; +import org.springframework.core.annotation.Order; +import org.springframework.stereotype.Component; + +/** + * Spring Batch 6.0's new {@link JobOperator#recover(JobExecution)} (see + * {@code docs/09-jobexecutionalreadyrunning-and-recover.md}): the fix for the stuck job this + * module's {@code reject} profile produces. A rejected partition submission leaves its + * {@code StepExecution} parked at {@code STARTING}/{@code EXECUTING} forever even though the + * manager step and the {@code JobExecution} both reach {@code FAILED} — and the orphaned + * child rows are exactly what makes {@code JobOperator#start} on the same identifying parameters + * throw {@code JobExecutionAlreadyRunningException} on every subsequent attempt. {@code recover} + * walks the execution's steps and force-closes anything still marked running, after which a + * normal restart proceeds. + * + *

Runs before {@link OrderIngestRunner} ({@code @Order(0)} vs. the default) when the + * {@code recover} profile is active, so a single JVM invocation both recovers and restarts. + */ +@Component +@Profile("recover") +@Order(0) +public class RecoveryRunner implements ApplicationRunner { + + private final JobOperator jobOperator; + private final JobExplorer jobExplorer; + + @Value("${recover.job-execution-id}") + private long jobExecutionId; + + public RecoveryRunner(JobOperator jobOperator, JobExplorer jobExplorer) { + this.jobOperator = jobOperator; + this.jobExplorer = jobExplorer; + } + + @Override + public void run(ApplicationArguments args) { + JobExecution execution = jobExplorer.getJobExecution(jobExecutionId); + if (execution == null) { + System.out.println("RECOVER: no JobExecution with id=" + jobExecutionId); + return; + } + System.out.println("RECOVER: before -> status=" + execution.getStatus()); + execution.getStepExecutions().forEach(se -> + System.out.println("RECOVER: step=" + se.getStepName() + " status=" + se.getStatus())); + JobExecution recovered = jobOperator.recover(execution); + System.out.println("RECOVER: after -> status=" + recovered.getStatus()); + recovered.getStepExecutions().forEach(se -> + System.out.println("RECOVER: step=" + se.getStepName() + " status=" + se.getStatus())); + } +} diff --git a/spring-batch-partitioning/src/main/java/com/ankurm/batchpartition/web/PartitionInsightController.java b/spring-batch-partitioning/src/main/java/com/ankurm/batchpartition/web/PartitionInsightController.java new file mode 100644 index 0000000..fa50f76 --- /dev/null +++ b/spring-batch-partitioning/src/main/java/com/ankurm/batchpartition/web/PartitionInsightController.java @@ -0,0 +1,41 @@ +package com.ankurm.batchpartition.web; + +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.PathVariable; +import org.springframework.web.bind.annotation.RestController; + +import java.util.List; +import java.util.Map; + +/** + * Prints the real thing instead of the remembered thing: which OS thread executed which + * partition of a given job execution, in what order, and for how long. Delete before shipping + * — see the Javadoc on {@link com.ankurm.batchpartition.partition.PartitionStatsListener}. + */ +@RestController +public class PartitionInsightController { + + private final JdbcTemplate jdbcTemplate; + + public PartitionInsightController(JdbcTemplate jdbcTemplate) { + this.jdbcTemplate = jdbcTemplate; + } + + @GetMapping("/batch/partitions/{jobExecutionId}") + public List> partitions(@PathVariable long jobExecutionId) { + return jdbcTemplate.queryForList( + "SELECT partition_name, thread_name, read_count, started_at, finished_at, duration_ms, exit_code " + + "FROM PARTITION_STATS WHERE job_execution_id = ? ORDER BY started_at", + jobExecutionId); + } + + @GetMapping("/batch/partitions/latest") + public List> latest() { + return jdbcTemplate.queryForList( + "SELECT job_execution_id, partition_name, thread_name, read_count, started_at, finished_at, " + + "duration_ms, exit_code FROM PARTITION_STATS " + + "WHERE job_execution_id = (SELECT MAX(job_execution_id) FROM PARTITION_STATS) " + + "ORDER BY started_at"); + } +} diff --git a/spring-batch-partitioning/src/main/resources/application.yml b/spring-batch-partitioning/src/main/resources/application.yml new file mode 100644 index 0000000..abb135d --- /dev/null +++ b/spring-batch-partitioning/src/main/resources/application.yml @@ -0,0 +1,36 @@ +spring: + batch: + job: + # Same reasoning as the spring-batch module: ImportRunner-style explicit launch, not the + # auto-configured JobLauncherApplicationRunner, which would launch every Job bean with + # parameterless defaults and run it 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 + +server: + port: 8081 + +partition: + shards-dir: ./data/shards + grid-size: 4 + pool-core-size: 4 + pool-max-size: 4 + pool-queue-capacity: 0 + +risk: + iterations: 150 + high-risk-amount-cents: 9500000 + +logging: + level: + org.springframework.batch: INFO diff --git a/spring-batch-partitioning/src/main/resources/schema.sql b/spring-batch-partitioning/src/main/resources/schema.sql new file mode 100644 index 0000000..c770245 --- /dev/null +++ b/spring-batch-partitioning/src/main/resources/schema.sql @@ -0,0 +1,24 @@ +CREATE TABLE IF NOT EXISTS ORDER_RISK_SUMMARY ( + order_id BIGINT PRIMARY KEY, + customer_id BIGINT NOT NULL, + amount_cents BIGINT NOT NULL, + region VARCHAR(16) NOT NULL, + risk_score INT NOT NULL, + high_risk BOOLEAN NOT NULL, + partition_name VARCHAR(64) NOT NULL +); + +-- Hidden runtime state this module makes visible: which thread actually executed which +-- partition, how many rows it read, and how long it took. See +-- web/PartitionInsightController and docs/05-the-diagnostic-endpoint.md. +CREATE TABLE IF NOT EXISTS PARTITION_STATS ( + id BIGINT AUTO_INCREMENT PRIMARY KEY, + job_execution_id BIGINT NOT NULL, + partition_name VARCHAR(64) NOT NULL, + thread_name VARCHAR(128) NOT NULL, + read_count BIGINT NOT NULL, + started_at TIMESTAMP NOT NULL, + finished_at TIMESTAMP NOT NULL, + duration_ms BIGINT NOT NULL, + exit_code VARCHAR(32) NOT NULL +); diff --git a/spring-batch-partitioning/src/test/java/com/ankurm/batchpartition/PartitionerGridSizeTest.java b/spring-batch-partitioning/src/test/java/com/ankurm/batchpartition/PartitionerGridSizeTest.java new file mode 100644 index 0000000..72d05d9 --- /dev/null +++ b/spring-batch-partitioning/src/test/java/com/ankurm/batchpartition/PartitionerGridSizeTest.java @@ -0,0 +1,50 @@ +package com.ankurm.batchpartition; + +import org.junit.jupiter.api.Test; +import org.springframework.batch.core.partition.support.MultiResourcePartitioner; +import org.springframework.batch.infrastructure.item.ExecutionContext; +import org.springframework.core.io.FileSystemResource; +import org.springframework.core.io.Resource; + +import java.io.IOException; +import java.nio.file.Files; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Pins the claim in {@code docs/03-what-gridsize-actually-controls.md}: decompiling + * {@code MultiResourcePartitioner.partition(int)} in {@code spring-batch-core-6.0.5.jar} shows + * the method never reads its {@code int gridSize} argument — it loops over the configured + * {@code resources} array (it does call {@code resource.getURL()} on each one, though, which is + * why this test uses real temp files rather than {@code ByteArrayResource}: the first draft did, + * and failed with {@code FileNotFoundException: Byte array resource cannot be resolved to URL} + * — itself a small, real, verified fact about what this method requires of its resources). + * If a future Spring Batch release changes the gridSize behaviour, this test is what breaks + * first, before an article claim goes stale silently. + */ +class PartitionerGridSizeTest { + + @Test + void partitionCountFollowsResourceCountNotGridSize() throws IOException { + var partitioner = new MultiResourcePartitioner(); + Resource[] resources = { + new FileSystemResource(Files.createTempFile("shard-one-", ".csv")), + new FileSystemResource(Files.createTempFile("shard-two-", ".csv")), + new FileSystemResource(Files.createTempFile("shard-three-", ".csv")), + }; + partitioner.setResources(resources); + + Map partitions = partitioner.partition(10); + + try (var t = new Transcript("02-gridsize-ignored.txt", + "MultiResourcePartitioner.partition(10) with 3 resources")) { + t.line("resources given: %d", resources.length); + t.line("gridSize argument passed to partition(): 10"); + t.line("partitions actually returned: %d", partitions.size()); + t.line("partition keys: %s", String.join(", ", partitions.keySet())); + } + + assertThat(partitions).hasSize(3); + } +} diff --git a/spring-batch-partitioning/src/test/java/com/ankurm/batchpartition/Transcript.java b/spring-batch-partitioning/src/test/java/com/ankurm/batchpartition/Transcript.java new file mode 100644 index 0000000..f19648d --- /dev/null +++ b/spring-batch-partitioning/src/test/java/com/ankurm/batchpartition/Transcript.java @@ -0,0 +1,52 @@ +package com.ankurm.batchpartition; + +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); + } +} diff --git a/spring-batch-partitioning/src/test/java/com/ankurm/batchpartition/processing/RiskScoringProcessorTest.java b/spring-batch-partitioning/src/test/java/com/ankurm/batchpartition/processing/RiskScoringProcessorTest.java new file mode 100644 index 0000000..930eeee --- /dev/null +++ b/spring-batch-partitioning/src/test/java/com/ankurm/batchpartition/processing/RiskScoringProcessorTest.java @@ -0,0 +1,39 @@ +package com.ankurm.batchpartition.processing; + +import com.ankurm.batchpartition.Transcript; +import com.ankurm.batchpartition.domain.Order; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * {@link RiskScoringProcessor} is pure and deterministic on purpose: {@code docs/06-why-cpu-bound-not-io-bound.md} + * leans on that determinism to prove two separate JVM runs of the same shard produce byte-identical + * {@code ORDER_RISK_SUMMARY} rows — a claim this test pins at the processor level before the + * full job ever runs. + */ +class RiskScoringProcessorTest { + + @Test + void sameInputAlwaysProducesSameScore() { + var processor = new RiskScoringProcessor(150, 9_500_000L); + var order = new Order(42L, 777L, 1_234_567L, "NORTH"); + + var first = processor.process(order); + var second = processor.process(order); + + try (var t = new Transcript("01-processor-determinism.txt", "RiskScoringProcessor determinism, threshold behaviour")) { + t.line("order: %s", order); + t.line("first.process() -> riskScore=%d highRisk=%b", first.riskScore(), first.highRisk()); + t.line("second.process() -> riskScore=%d highRisk=%b", second.riskScore(), second.highRisk()); + t.section("amount threshold, score held constant by a low iteration count"); + var below = processor.process(new Order(1L, 1L, 9_499_999L, "EAST")); + var above = processor.process(new Order(1L, 1L, 9_500_001L, "EAST")); + t.line("amountCents=9,499,999 (at threshold, exclusive) -> highRisk=%b", below.highRisk()); + t.line("amountCents=9,500,001 (over threshold) -> highRisk=%b", above.highRisk()); + } + + assertThat(first.riskScore()).isEqualTo(second.riskScore()); + assertThat(first.highRisk()).isEqualTo(second.highRisk()); + } +}