spring-batch

Spring Batch

Spring Boot 3.x には Spring Batch 5 が同梱されています。API は 4.x から大幅に変更されており、オンラインのサンプルの大半は誤っています。エージェントが生成するコードで特に問題を引き起こしやすいルールは、次の 2 つです。

1. @EnableBatchProcessing は追加しないでください。 Boot が JobRepository、JobLauncher、トランザクションマネージャーを自動構成します。@EnableBatchProcessing を追加すると、その自動構成が無効になり、配線済みのすべての Bean が使えなくなります。
2. JobBuilderFactory と StepBuilderFactory は廃止されています。 Boot がすでに作成した JobRepository と PlatformTransactionManager を注入し、new JobBuilder(name, jobRepository)／new StepBuilder(name, jobRepository) でジョブとステップを構築してください。

Job と Step（Spring Batch 5 API）

@Configuration
@RequiredArgsConstructor
public class OrderExportJobConfig {

    @Bean
    public Job orderExportJob(JobRepository jobRepository, Step exportStep) {
        return new JobBuilder("orderExportJob", jobRepository)
            .incrementer(new RunIdIncrementer()) // lets the same job be re-run; see "Idempotency"
            .start(exportStep)
            .build();
    }

    @Bean
    public Step exportStep(JobRepository jobRepository,
                           PlatformTransactionManager txManager, // Boot's, injected — do NOT new one up
                           ItemReader<Order> reader,
                           ItemProcessor<Order, OrderRow> processor,
                           ItemWriter<OrderRow> writer) {
        return new StepBuilder("exportStep", jobRepository)
            .<Order, OrderRow>chunk(500, txManager) // chunk size is the commit interval — and a TX boundary
            .reader(reader)
            .processor(processor)
            .writer(writer)
            .faultTolerant()
            .skip(FlatFileParseException.class)
            .skipLimit(50)
            .build();
    }
}

chunk(500, txManager) は、500 件のアイテムを読み込み、それぞれを処理し、500 件のリストをライターに渡して、1 つのトランザクションをコミットし、これを繰り返すという意味です。チャンクは再開の単位であり、ロールバックの単位でもあります。

冪等性と再開可能性 — 運用上の最大の落とし穴

JobInstance は、識別対象の JobParameters によって識別されます。同じ識別対象のパラメーターで同じジョブを 2 回起動すると、次のようになります。

JobInstanceAlreadyCompleteException: A job instance already exists and is complete

これは仕様どおりの動作です。Batch は完了済みの処理を再実行しません。この場合の対処方法は 2 つあります。

// Option A — RunIdIncrementer on the job (above) + JobLauncherApplicationRunner bumps run.id each launch.
// Option B — add a unique identifying parameter yourself when launching:
JobParameters params = new JobParametersBuilder()
    .addString("status", "COMPLETED")              // identifying — part of the instance key
    .addLong("run.id", System.currentTimeMillis()) // identifying & unique — makes each run a new instance
    .toJobParameters();

インスタンスの識別に影響させたくないメタデータ（ログには記録するものの、キーには使わないリクエスト ID など）には、false フラグを指定してパラメーターを非識別対象にします。

.addString("requestId", requestId, false) // non-identifying — excluded from the instance key

一方、失敗したジョブを同じパラメーターで再起動すると、ジョブは再開されます。完了済みのステップはスキップされ、失敗したステップは最後にコミットされたチャンクから再開されます。これがメタデータテーブルの役割です。失敗時に再開したい場合は、常に一意のパラメーターを渡すことで、この機能を無効にしないでください。

ItemReader — ソートキーとスレッドセーフ性

@Bean
@StepScope // required: late-binds jobParameters at step execution, not context startup
public JpaPagingItemReader<Order> orderReader(
        EntityManagerFactory emf,
        @Value("#{jobParameters['status']}") String status) {
    return new JpaPagingItemReaderBuilder<Order>()
        .name("orderReader")
        .entityManagerFactory(emf)
        .queryString("SELECT o FROM Order o WHERE o.status = :status ORDER BY o.id") // ORDER BY is MANDATORY
        .parameterValues(Map.of("status", OrderStatus.valueOf(status)))
        .pageSize(500) // keep pageSize == chunk size
        .build();
}

• ページングリーダーでは、一意のカラムによる決定的な ORDER BY が必要です。 これがないと、DB はページごとに任意の順序で行を返すため、行がスキップされたり、二重に処理されたりします。これはエラーではなく、気付かないままデータが破損する問題です。
• JdbcCursorItemReader はスレッドセーフではありません。 JdbcPagingItemReader／JpaPagingItemReader はマルチスレッドのステップでも安全に使用できます。マルチスレッドのステップでスレッドセーフではないリーダーを使う場合は、SynchronizedItemStreamReader でラップしてください。
• 同じジョブ内で、ページングに使うカラムを変更しないでください。 リーダーが WHERE status = 'PENDING' ORDER BY id でページングしている間に、ライターが status を PENDING から DONE に変更すると、結果セットの内容が途中で変化し、ページが欠落します。安定したスナップショットを読み込む、変更されない id でページングする、またはカーソルリーダーを使ってください。

ItemProcessor — null を返すとフィルタリング

@Component
public class OrderProcessor implements ItemProcessor<Order, OrderRow> {
    @Override
    public OrderRow process(Order order) {
        if (order.getTotal().isZero()) {
            return null; // ⚠️ null = FILTER this item; it is NOT written and NOT an error
        }
        return OrderRow.from(order);
    }
}

null を返すと、そのアイテムはチャンクから通知なく除外されます。これはフィルタリング機能としては便利ですが、通過させるつもりで誤って null を返した場合には危険です。

ItemWriter — Spring Batch 5 でのシグネチャ変更

5.x では、ライターが受け取るのは List<? extends T> ではなく、Chunk<? extends T> です。

@Override
public void write(Chunk<? extends OrderRow> chunk) { // was List<? extends T> in 4.x
    repository.saveAll(chunk.getItems());
}

SQL 書き込みには、行ごとの保存よりも、バッチ処理に対応した JDBC ライターを優先してください。addBatch() を 1 回使用できます。

@Bean
public JdbcBatchItemWriter<OrderRow> orderWriter(DataSource dataSource) {
    return new JdbcBatchItemWriterBuilder<OrderRow>()
        .dataSource(dataSource)
        .sql("INSERT INTO order_export (id, total) VALUES (:id, :total)")
        .beanMapped()
        .build();
}

ライターはチャンクのトランザクション内で実行されます。ライターからメール送信、Kafka へのメッセージ発行、Webhook の呼び出しを行わないでください。チャンクがロールバックされたとしても、これらはすでに実行されています。代わりに、副作用の実行をジョブ完了時に紐付けてください（[[transactional-patterns]] と、以下のリスナーを参照してください）。

ジョブの起動

Boot はデフォルトで、起動時にすべての Job Bean を実行します。スケジュール起動やオンデマンド起動を行う場合は、これを無効にし、明示的に起動してください。

spring:
  batch:
    job:
      enabled: false           # don't run jobs on app startup; we trigger them ourselves
    jdbc:
      initialize-schema: never # manage BATCH_* tables with Flyway in prod (see below)

@Component
@RequiredArgsConstructor
public class OrderExportScheduler {

    private final JobLauncher jobLauncher;
    private final Job orderExportJob;

    @Scheduled(cron = "0 0 2 * * *")
    public void runNightly() throws JobExecutionException {
        JobParameters params = new JobParametersBuilder()
            .addString("status", "COMPLETED")
            .addLong("run.id", System.currentTimeMillis())
            .toJobParameters();
        jobLauncher.run(orderExportJob, params);
    }
}

デフォルトの JobLauncher は同期型（SyncTaskExecutor）です。jobLauncher.run(...) はジョブ全体が完了するまで @Scheduled のスレッドをブロックします。バックグラウンドで実行したい場合は、非同期の TaskExecutor を設定したランチャーを使うか、ブロックを許容できる場合に限り、リクエストのスレッドから起動してください。

本番環境でのメタデータスキーマ

Spring Batch では、BATCH_JOB_INSTANCE、BATCH_JOB_EXECUTION、BATCH_STEP_EXECUTION、… テーブルが必要です。 開発環境や組み込み DB では initialize-schema: always で問題ありませんが、Batch に起動時の本番データベースへの DDL 実行を許可しないでください。initialize-schema: never を設定し、バージョン管理された [[flyway-migrations]] マイグレーションとしてスキーマを用意してください（正規の DDL は spring-batch-core 内の org/springframework/batch/core/schema-*.sql にあります）。

リスナー — チャンクごとではなくジョブの後に実行する必要がある処理

@Bean
public Job orderExportJob(JobRepository jobRepository, Step exportStep) {
    return new JobBuilder("orderExportJob", jobRepository)
        .incrementer(new RunIdIncrementer())
        .listener(new JobExecutionListener() {
            @Override public void afterJob(JobExecution exec) {
                if (exec.getStatus() == BatchStatus.COMPLETED) {
                    notifier.notifyExportReady(exec.getJobParameters()); // safe: all chunks committed
                }
            }
        })
        .start(exportStep)
        .build();
}

必要がなければ Batch を使わない

Spring Batch の複雑さ（メタデータテーブル、再起動、チャンク化）が活きるのは、大規模で、再起動可能かつ監査可能な一括ジョブです。手早く一度だけ実行する非同期タスクには、@Async や @Scheduled ループのほうが軽量です。リトライを伴う永続的なバックグラウンドジョブには、ジョブキューのほうが適しています。規模に合ったツールを選んでください。

注意点

• エージェントが @EnableBatchProcessing を追加する — Boot 3 では自動構成が無効になるため、削除して JobRepository をインジェクトします
• エージェントが JobBuilderFactory / StepBuilderFactory を使用する — Batch 5 で削除されました。new JobBuilder(name, repo) / new StepBuilder(name, repo) を使用します
• エージェントがトランザクションマネージャーなしで .chunk(500) を呼び出す — Batch 5 では .chunk(500, txManager) が必要です
• エージェントが write(List<? extends T> items) を記述する — Batch 5 のシグネチャは write(Chunk<? extends T> chunk) です
• エージェントが同じパラメーターでジョブを再実行し、JobInstanceAlreadyCompleteException が発生する — RunIdIncrementer または一意な識別パラメーターを追加します
• エージェントが失敗時に再開すべきジョブに、実行ごとに一意なパラメーターを追加する — 再起動可能性が損なわれます。新しいインスタンスを開始するときだけ追加します
• エージェントが ORDER BY のない（または一意でない）ページングリーダーのクエリを記述する — ページの処理で行が暗黙的にスキップまたは重複します。一意な列で並べ替えます
• エージェントがマルチスレッドステップで JdbcCursorItemReader を使用する — スレッドセーフではありません。ページングリーダーまたは SynchronizedItemStreamReader を使用します
• エージェントが同じジョブ内でライターが変更する列をページングに使用する — 結果セットがずれます。不変の ID でページングします
• エージェントがパススルーを想定するプロセッサーから null を返す — null はアイテムをフィルター処理（破棄）します
• エージェントが ItemWriter からメールを送信する／イベントを発行する — チャンク TX 内で実行されます。afterJob リスナーで実行します
• エージェントが jobParameters を読み取るリーダーに @StepScope を付け忘れる — @Value("#{jobParameters[...]}") がバインドされるのはステップスコープ内だけです
• エージェントが Web アプリの起動時にジョブを実行したままにする — spring.batch.job.enabled=false を設定し、明示的に起動します
• エージェントが initialize-schema: always による本番 DB への DDL 実行を許可する — never を使用し、BATCH_* テーブル用の Flyway マイグレーションを追加します