Add spring-batch-partitioning: manager/worker partitioning, gridSize semantics, rejected-partition recovery, and a real 10M-row scaling sweep
This commit is contained in:
@@ -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 <dir> <rows> <shards> [--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).
|
||||
@@ -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.
|
||||
|
||||
<figure>
|
||||
<svg viewBox="0 0 740 300" role="img" aria-label="One manager step asks a Partitioner for four ExecutionContexts, one per shard file, and hands each to an identical worker step running on its own thread; all four worker steps write to the same ORDER_RISK_SUMMARY table.">
|
||||
<style>.h{font:600 12px sans-serif;fill:#1a1a1a}.c{font:11px sans-serif;fill:#4b5563}.m{font:11px monospace;fill:#1a1a1a}</style>
|
||||
<rect x="300" y="16" width="160" height="40" rx="4" fill="#e8eefc" stroke="#5b7fc7"/>
|
||||
<text x="326" y="41" class="h">ordersManagerStep</text>
|
||||
<text x="130" y="90" class="c">Partitioner.partition(gridSize) returns one ExecutionContext per shard file</text>
|
||||
<rect x="20" y="110" width="160" height="36" rx="4" fill="#f4f5f7" stroke="#b7bec9"/><text x="40" y="133" class="m">fileName=shard-00.csv</text>
|
||||
<rect x="200" y="110" width="160" height="36" rx="4" fill="#f4f5f7" stroke="#b7bec9"/><text x="220" y="133" class="m">fileName=shard-01.csv</text>
|
||||
<rect x="380" y="110" width="160" height="36" rx="4" fill="#f4f5f7" stroke="#b7bec9"/><text x="400" y="133" class="m">fileName=shard-02.csv</text>
|
||||
<rect x="560" y="110" width="160" height="36" rx="4" fill="#f4f5f7" stroke="#b7bec9"/><text x="580" y="133" class="m">fileName=shard-03.csv</text>
|
||||
<rect x="20" y="180" width="160" height="60" rx="4" fill="#e7f4ea" stroke="#4a9d63"/><text x="40" y="205" class="h">worker step</text><text x="40" y="222" class="c">partition0, thread A</text>
|
||||
<rect x="200" y="180" width="160" height="60" rx="4" fill="#e7f4ea" stroke="#4a9d63"/><text x="220" y="205" class="h">worker step</text><text x="220" y="222" class="c">partition1, thread B</text>
|
||||
<rect x="380" y="180" width="160" height="60" rx="4" fill="#e7f4ea" stroke="#4a9d63"/><text x="400" y="205" class="h">worker step</text><text x="400" y="222" class="c">partition2, thread C</text>
|
||||
<rect x="560" y="180" width="160" height="60" rx="4" fill="#e7f4ea" stroke="#4a9d63"/><text x="580" y="205" class="h">worker step</text><text x="580" y="222" class="c">partition3, thread D</text>
|
||||
<rect x="260" y="268" width="220" height="24" rx="4" fill="#fdeccf" stroke="#c9973f"/>
|
||||
<text x="275" y="285" class="c">ORDER_RISK_SUMMARY (shared H2 file)</text>
|
||||
</svg>
|
||||
</figure>
|
||||
|
||||
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)
|
||||
@@ -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<Order> shardReader, ItemProcessor<Order, RiskScoredOrder> riskProcessor,
|
||||
ItemWriter<RiskScoredOrder> riskWriter, PartitionStatsListener partitionStatsListener) {
|
||||
return new StepBuilder("ordersWorkerStep", jobRepository)
|
||||
.<Order, RiskScoredOrder>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<Order> shardReader(@Value("#{stepExecutionContext['fileName']}") Resource shardFile) {
|
||||
return new FlatFileItemReaderBuilder<Order>()
|
||||
.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)
|
||||
@@ -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)
|
||||
@@ -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<RiskScoredOrder> riskWriter(JdbcTemplate jdbcTemplate,
|
||||
@Value("#{stepExecution.stepName}") String partitionName) {
|
||||
return new JdbcBatchItemWriterBuilder<RiskScoredOrder>()
|
||||
.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)
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -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.
|
||||
|
||||
<figure>
|
||||
<svg viewBox="0 0 740 220" role="img" aria-label="Four partitions submitted to a pool of size one with a zero-capacity queue and AbortPolicy: partition1 runs and completes, the other three are rejected on submission and their StepExecutions are left at STARTING forever.">
|
||||
<style>.h{font:600 12px sans-serif;fill:#1a1a1a}.c{font:11px sans-serif;fill:#4b5563}.m{font:11px monospace;fill:#1a1a1a}</style>
|
||||
<rect x="20" y="20" width="160" height="50" rx="4" fill="#e7f4ea" stroke="#4a9d63"/><text x="35" y="42" class="h">partition1</text><text x="35" y="58" class="c">runs, COMPLETED</text>
|
||||
<rect x="200" y="20" width="160" height="50" rx="4" fill="#f7d9d3" stroke="#c56a54"/><text x="215" y="42" class="h">partition0</text><text x="215" y="58" class="c">rejected on submit</text>
|
||||
<rect x="380" y="20" width="160" height="50" rx="4" fill="#f7d9d3" stroke="#c56a54"/><text x="395" y="42" class="h">partition2</text><text x="395" y="58" class="c">rejected on submit</text>
|
||||
<rect x="560" y="20" width="160" height="50" rx="4" fill="#f7d9d3" stroke="#c56a54"/><text x="575" y="42" class="h">partition3</text><text x="575" y="58" class="c">rejected on submit</text>
|
||||
<text x="20" y="105" class="c">StepExecution status for the three rejected partitions:</text>
|
||||
<rect x="20" y="120" width="700" height="34" rx="4" fill="#fdeccf" stroke="#c9973f"/>
|
||||
<text x="34" y="142" class="m">STARTING / EXECUTING -- forever. No callback ever closes them out.</text>
|
||||
<text x="20" y="185" class="c">The manager step and JobExecution both reach FAILED cleanly. These three do not.</text>
|
||||
</svg>
|
||||
</figure>
|
||||
|
||||
## 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)
|
||||
@@ -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.
|
||||
|
||||
<figure>
|
||||
<svg viewBox="0 0 740 200" role="img" aria-label="JobExecution 1: partition1 completes, the other three fail. JobExecution 33, the restart: only partition0, partition2 and partition3 run again; partition1 is skipped entirely, its 5000-row result untouched.">
|
||||
<style>.h{font:600 12px sans-serif;fill:#1a1a1a}.c{font:11px sans-serif;fill:#4b5563}.m{font:11px monospace;fill:#1a1a1a}</style>
|
||||
<text x="20" y="24" class="h">JobExecution 1</text>
|
||||
<rect x="20" y="36" width="160" height="40" rx="4" fill="#e7f4ea" stroke="#4a9d63"/><text x="35" y="60" class="m">partition1: COMPLETED</text>
|
||||
<rect x="200" y="36" width="160" height="40" rx="4" fill="#f7d9d3" stroke="#c56a54"/><text x="215" y="60" class="m">partition0: FAILED</text>
|
||||
<rect x="380" y="36" width="160" height="40" rx="4" fill="#f7d9d3" stroke="#c56a54"/><text x="395" y="60" class="m">partition2: FAILED</text>
|
||||
<rect x="560" y="36" width="160" height="40" rx="4" fill="#f7d9d3" stroke="#c56a54"/><text x="575" y="60" class="m">partition3: FAILED</text>
|
||||
<text x="20" y="112" class="h">JobExecution 33 (restart, same shardsDir)</text>
|
||||
<rect x="20" y="124" width="160" height="40" rx="4" fill="#f4f5f7" stroke="#b7bec9" stroke-dasharray="3 3"/><text x="34" y="148" class="c">partition1: SKIPPED</text>
|
||||
<rect x="200" y="124" width="160" height="40" rx="4" fill="#e7f4ea" stroke="#4a9d63"/><text x="220" y="148" class="m">partition0: reran</text>
|
||||
<rect x="380" y="124" width="160" height="40" rx="4" fill="#e7f4ea" stroke="#4a9d63"/><text x="400" y="148" class="m">partition2: reran</text>
|
||||
<rect x="560" y="124" width="160" height="40" rx="4" fill="#e7f4ea" stroke="#4a9d63"/><text x="580" y="148" class="m">partition3: reran</text>
|
||||
<text x="20" y="185" class="c">manager READ_COUNT for execution 33 = 15000 (3 x 5000), not 20000</text>
|
||||
</svg>
|
||||
</figure>
|
||||
|
||||
## 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)
|
||||
@@ -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)
|
||||
@@ -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.
|
||||
|
||||
<figure>
|
||||
<svg viewBox="0 0 740 260" role="img" aria-label="Speedup versus grid size, two data volumes overlaid. At 10 million rows, speedup peaks at grid size 2 and degrades gently. At 300 thousand rows, it peaks lower and grid size 8 falls below 1.0 -- slower than not partitioning.">
|
||||
<style>.h{font:600 12px sans-serif;fill:#1a1a1a}.c{font:11px sans-serif;fill:#4b5563}.m{font:11px monospace;fill:#1a1a1a}</style>
|
||||
<line x1="60" y1="220" x2="700" y2="220" stroke="#b7bec9"/>
|
||||
<line x1="60" y1="220" x2="60" y2="20" stroke="#b7bec9"/>
|
||||
<text x="20" y="30" class="c">speedup</text>
|
||||
<text x="660" y="240" class="c">gridSize</text>
|
||||
<text x="55" y="115" class="c" text-anchor="end">1.0x</text>
|
||||
<line x1="60" y1="112" x2="700" y2="112" stroke="#e2e5ea" stroke-dasharray="3 3"/>
|
||||
<!-- 10M line: 1.00,1.42,1.33,1.27 mapped roughly -->
|
||||
<polyline points="100,112 260,44 420,60 580,72" fill="none" stroke="#5b7fc7" stroke-width="3"/>
|
||||
<circle cx="100" cy="112" r="4" fill="#5b7fc7"/><circle cx="260" cy="44" r="4" fill="#5b7fc7"/><circle cx="420" cy="60" r="4" fill="#5b7fc7"/><circle cx="580" cy="72" r="4" fill="#5b7fc7"/>
|
||||
<text x="600" y="66" class="c" fill="#5b7fc7">10M rows</text>
|
||||
<!-- 300K line: 1.00,1.10,0.99,0.75 -->
|
||||
<polyline points="100,112 260,96 420,114 580,168" fill="none" stroke="#c56a54" stroke-width="3"/>
|
||||
<circle cx="100" cy="112" r="4" fill="#c56a54"/><circle cx="260" cy="96" r="4" fill="#c56a54"/><circle cx="420" cy="114" r="4" fill="#c56a54"/><circle cx="580" cy="168" r="4" fill="#c56a54"/>
|
||||
<text x="600" y="180" class="c" fill="#c56a54">300K rows</text>
|
||||
<text x="95" y="235" class="m">1</text><text x="255" y="235" class="m">2</text><text x="415" y="235" class="m">4</text><text x="575" y="235" class="m">8</text>
|
||||
</svg>
|
||||
</figure>
|
||||
|
||||
## 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)
|
||||
@@ -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)
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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<java.lang.String, org.springframework.batch.infrastructure.item.ExecutionContext> partition(int);
|
||||
}
|
||||
|
||||
Note the return type: Map<String, org.springframework.batch.infrastructure.item.ExecutionContext>.
|
||||
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)
|
||||
@@ -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.
|
||||
@@ -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"
|
||||
}
|
||||
]
|
||||
@@ -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.
|
||||
@@ -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.
|
||||
@@ -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.
|
||||
@@ -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.
|
||||
@@ -0,0 +1,76 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<parent>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-parent</artifactId>
|
||||
<version>4.1.1</version>
|
||||
<relativePath/>
|
||||
</parent>
|
||||
|
||||
<groupId>com.ankurm</groupId>
|
||||
<artifactId>spring-batch-partitioning</artifactId>
|
||||
<version>1.0.0</version>
|
||||
<name>spring-batch-partitioning</name>
|
||||
<description>Spring Batch partitioning and parallel steps: scaling a 10-million-row job</description>
|
||||
|
||||
<properties>
|
||||
<java.version>25</java.version>
|
||||
</properties>
|
||||
|
||||
<dependencies>
|
||||
<!-- Brings spring-boot-starter-batch (BatchAutoConfiguration) plus
|
||||
spring-boot-starter-batch-jdbc (a real JDBC-backed JobRepository, required here: the
|
||||
whole point of the restart demo in docs/07-restart-and-partitions.md is that the
|
||||
manager step's bookkeeping about which partitions already completed survives the JVM
|
||||
exiting). -->
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-batch-jdbc</artifactId>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-jdbc</artifactId>
|
||||
</dependency>
|
||||
|
||||
<!-- The diagnostic endpoint in web/PartitionInsightController -->
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-web</artifactId>
|
||||
</dependency>
|
||||
|
||||
<!-- File-based H2: the PARTITION_STATS and ORDER_RISK_SUMMARY tables, and the job
|
||||
repository itself, all need to survive one java -jar exiting and another starting for
|
||||
the restart-only-the-failed-partition demonstration to mean anything. -->
|
||||
<dependency>
|
||||
<groupId>com.h2database</groupId>
|
||||
<artifactId>h2</artifactId>
|
||||
<scope>runtime</scope>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-test</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.springframework.batch</groupId>
|
||||
<artifactId>spring-batch-test</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
<build>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-maven-plugin</artifactId>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</build>
|
||||
</project>
|
||||
+61
@@ -0,0 +1,61 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Generates the sharded order CSVs this module partitions over.
|
||||
|
||||
Usage:
|
||||
generate-shards.py <output-dir> <total-rows> <num-shards> [--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()
|
||||
Executable
+28
@@ -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 <db-name> <shards-dir> [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 <db-name> <shards-dir> [grid-size] [pool-size] [extra args...]}"
|
||||
SHARDS_DIR="${2:?usage: run.sh <db-name> <shards-dir> [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"
|
||||
+11
@@ -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);
|
||||
}
|
||||
}
|
||||
+257
@@ -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}.
|
||||
*
|
||||
* <p>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<Order> shardReader(@Value("#{stepExecutionContext['fileName']}") Resource shardFile) {
|
||||
return new FlatFileItemReaderBuilder<Order>()
|
||||
.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<Order, RiskScoredOrder> 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()}.
|
||||
*
|
||||
* <p>{@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<RiskScoredOrder> riskWriter(JdbcTemplate jdbcTemplate,
|
||||
@Value("#{stepExecution.stepName}") String partitionName) {
|
||||
return new JdbcBatchItemWriterBuilder<RiskScoredOrder>()
|
||||
.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<Order> shardReader, ItemProcessor<Order, RiskScoredOrder> riskProcessor,
|
||||
ItemWriter<RiskScoredOrder> riskWriter, PartitionStatsListener partitionStatsListener) {
|
||||
return new StepBuilder("ordersWorkerStep", jobRepository)
|
||||
.<Order, RiskScoredOrder>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).
|
||||
*
|
||||
* <p>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();
|
||||
}
|
||||
}
|
||||
@@ -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) {
|
||||
}
|
||||
+5
@@ -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) {
|
||||
}
|
||||
+53
@@ -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<LocalDateTime> 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();
|
||||
}
|
||||
}
|
||||
+37
@@ -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<Order, RiskScoredOrder> {
|
||||
|
||||
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);
|
||||
}
|
||||
}
|
||||
+47
@@ -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());
|
||||
}
|
||||
}
|
||||
}
|
||||
+58
@@ -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.
|
||||
*
|
||||
* <p>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()));
|
||||
}
|
||||
}
|
||||
+41
@@ -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<Map<String, Object>> 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<Map<String, Object>> 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");
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
@@ -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
|
||||
);
|
||||
+50
@@ -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<String, ExecutionContext> 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);
|
||||
}
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
+39
@@ -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());
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user