Spring Batch

@Bean
public Step invoiceStep(JobRepository jobRepository,
                        PlatformTransactionManager txManager) {
    return new StepBuilder("invoiceStep", jobRepository)
        .<RawInvoice, ProcessedInvoice>chunk(100, txManager)  // commit-interval = 100
        .reader(invoiceReader())
        .processor(invoiceProcessor())
        .writer(invoiceWriter())
        .faultTolerant()
        .skipLimit(10)
        .skip(InvalidRecordException.class)
        .retryLimit(3)
        .retry(TransientDataAccessException.class)
        .build();
}

@Bean
public Step cleanupStep(JobRepository jobRepository,
                        PlatformTransactionManager txManager) {
    return new StepBuilder("cleanupStep", jobRepository)
        .tasklet((contribution, chunkContext) -> {
            // Delete temp files older than 7 days
            FileUtils.cleanDirectory(tempDir);
            return RepeatStatus.FINISHED;
        }, txManager)
        .build();
}

@Bean
public FlatFileItemReader<Invoice> csvReader() {
    return new FlatFileItemReaderBuilder<Invoice>()
        .name("invoiceCsvReader")
        .resource(new ClassPathResource("invoices.csv"))
        .delimited()
        .names("id", "customerId", "amount", "dueDate")
        .targetType(Invoice.class)
        .linesToSkip(1)  // skip header
        .build();
}

@Bean
public JdbcCursorItemReader<Invoice> cursorReader(DataSource dataSource) {
    return new JdbcCursorItemReaderBuilder<Invoice>()
        .name("invoiceCursorReader")
        .dataSource(dataSource)
        .sql("SELECT id, customer_id, amount, status FROM invoices WHERE status = 'PENDING'")
        .rowMapper(new BeanPropertyRowMapper<>(Invoice.class))
        .fetchSize(500)       // JDBC fetch size (not chunk size)
        .saveState(true)      // persist cursor position for restart
        .build();
}

@Bean
public JpaPagingItemReader<Invoice> jpaPagingReader(EntityManagerFactory emf) {
    return new JpaPagingItemReaderBuilder<Invoice>()
        .name("invoiceJpaReader")
        .entityManagerFactory(emf)
        .queryString("SELECT i FROM Invoice i WHERE i.status = :status")
        .parameterValues(Map.of("status", "PENDING"))
        .pageSize(200)        // items per page (usually = chunk size)
        .saveState(true)
        .build();
}

@Bean
public FlatFileItemWriter<ProcessedInvoice> csvWriter() {
    return new FlatFileItemWriterBuilder<ProcessedInvoice>()
        .name("invoiceCsvWriter")
        .resource(new FileSystemResource("output/processed-invoices.csv"))
        .delimited()
        .names("id", "customerId", "amount", "processedDate")
        .headerCallback(writer -> writer.write("ID,Customer,Amount,Processed"))
        .build();
}

@Bean
public JdbcBatchItemWriter<ProcessedInvoice> jdbcWriter(DataSource dataSource) {
    return new JdbcBatchItemWriterBuilder<ProcessedInvoice>()
        .dataSource(dataSource)
        .sql("INSERT INTO processed_invoices (id, customer_id, amount, status) " +
             "VALUES (:id, :customerId, :amount, :status)")
        .beanMapped()         // uses BeanPropertySqlParameterSource
        .assertUpdates(true)  // fail if expected row count doesn't match
        .build();
}

@Bean
public JpaItemWriter<ProcessedInvoice> jpaWriter(EntityManagerFactory emf) {
    JpaItemWriter<ProcessedInvoice> writer = new JpaItemWriter<>();
    writer.setEntityManagerFactory(emf);
    writer.setUsePersist(true);  // persist vs merge
    return writer;
}

@Bean
public Step resilientStep(JobRepository jobRepository,
                          PlatformTransactionManager txManager) {
    return new StepBuilder("resilientStep", jobRepository)
        .<Input, Output>chunk(100, txManager)
        .reader(reader())
        .processor(processor())
        .writer(writer())
        .faultTolerant()
        // Skip configuration
        .skipLimit(50)                              // max total skips
        .skip(ValidationException.class)            // skip this exception
        .skip(FlatFileParseException.class)
        .noSkip(DatabaseException.class)            // never skip this
        // Custom skip policy for complex logic
        .skipPolicy(new CustomSkipPolicy())
        .listener(new SkipListener<Input, Output>() {
            @Override
            public void onSkipInRead(Throwable t) {
                log.warn("Skipped record during read: {}", t.getMessage());
            }
            @Override
            public void onSkipInProcess(Input item, Throwable t) {
                log.warn("Skipped item {}: {}", item.getId(), t.getMessage());
                deadLetterService.save(item, t);
            }
        })
        .build();
}

@Bean
public Step retryStep(JobRepository jobRepository,
                      PlatformTransactionManager txManager) {
    return new StepBuilder("retryStep", jobRepository)
        .<Input, Output>chunk(100, txManager)
        .reader(reader())
        .processor(processor())
        .writer(writer())
        .faultTolerant()
        // Retry configuration
        .retryLimit(3)
        .retry(TransientDataAccessException.class)
        .retry(OptimisticLockingFailureException.class)
        .noRetry(NonTransientDataAccessException.class)
        // Backoff policy
        .backOffPolicy(new ExponentialBackOffPolicy() {{
            setInitialInterval(1000);   // 1s
            setMultiplier(2.0);         // 1s → 2s → 4s
            setMaxInterval(10000);      // cap at 10s
        }})
        .listener(new RetryListener() {
            @Override
            public <T, E extends Throwable> void onError(
                    RetryContext context, RetryCallback<T, E> callback, Throwable t) {
                log.warn("Retry attempt #{} failed: {}",
                    context.getRetryCount(), t.getMessage());
            }
        })
        .build();
}

@Bean
public Job billingJob(JobRepository jobRepository, Step invoiceStep, Step reportStep) {
    return new JobBuilder("billingJob", jobRepository)
        .start(invoiceStep)
        .next(reportStep)
        .preventRestart()      // set to prevent restart (default: restartable)
        .incrementer(new RunIdIncrementer())  // allows re-running same params
        .listener(new JobExecutionListener() {
            @Override
            public void afterJob(JobExecution jobExecution) {
                if (jobExecution.getStatus() == BatchStatus.COMPLETED) {
                    notificationService.sendSuccess();
                }
            }
        })
        .build();
}

// Partitioner — splits work into ranges
@Bean
public Partitioner invoicePartitioner(DataSource dataSource) {
    ColumnRangePartitioner partitioner = new ColumnRangePartitioner();
    partitioner.setColumn("id");
    partitioner.setTable("invoices");
    partitioner.setDataSource(dataSource);
    return partitioner;
}

// Master Step
@Bean
public Step masterStep(JobRepository jobRepository,
                       Step slaveStep,
                       Partitioner partitioner) {
    return new StepBuilder("masterStep", jobRepository)
        .partitioner("slaveStep", partitioner)
        .step(slaveStep)
        .gridSize(8)                          // number of partitions
        .taskExecutor(new SimpleAsyncTaskExecutor())  // parallel execution
        .build();
}

// Slave Step — each partition runs independently
@Bean
@StepScope
public JdbcCursorItemReader<Invoice> partitionedReader(
        @Value("#{stepExecutionContext['minValue']}") Long minId,
        @Value("#{stepExecutionContext['maxValue']}") Long maxId,
        DataSource dataSource) {
    return new JdbcCursorItemReaderBuilder<Invoice>()
        .name("partitionedReader")
        .dataSource(dataSource)
        .sql("SELECT * FROM invoices WHERE id BETWEEN ? AND ?")
        .preparedStatementSetter(ps -> {
            ps.setLong(1, minId);
            ps.setLong(2, maxId);
        })
        .rowMapper(new BeanPropertyRowMapper<>(Invoice.class))
        .build();
}

@Component
@RequiredArgsConstructor
public class BatchScheduler {

    private final JobLauncher jobLauncher;
    private final Job billingJob;

    @Scheduled(cron = "0 0 2 * * *")  // Every day at 2 AM
    public void runBillingJob() throws Exception {
        JobParameters params = new JobParametersBuilder()
            .addLocalDate("runDate", LocalDate.now())
            .addLong("runId", System.currentTimeMillis())  // unique per run
            .toJobParameters();

        JobExecution execution = jobLauncher.run(billingJob, params);
        log.info("Job finished with status: {}", execution.getStatus());
    }
}

@Configuration
public class QuartzBatchConfig {

    @Bean
    public JobDetail billingJobDetail() {
        return JobBuilder.newJob(BatchJobLauncherDetail.class)
            .withIdentity("billingJob")
            .storeDurably()
            .usingJobData("jobName", "billingJob")
            .build();
    }

    @Bean
    public Trigger billingTrigger(JobDetail billingJobDetail) {
        return TriggerBuilder.newTrigger()
            .forJob(billingJobDetail)
            .withIdentity("billingTrigger")
            .withSchedule(CronScheduleBuilder
                .cronSchedule("0 0 2 * * ?")
                .withMisfireHandlingInstructionFireAndProceed())
            .build();
    }
}

@Configuration
@EnableBatchProcessing
public class BatchConfig {

    @Bean
    public Job billingJob(JobRepository jobRepository,
                          Step processStep,
                          Step cleanupStep) {
        return new JobBuilder("billingJob", jobRepository)
            .start(processStep)
            .next(cleanupStep)
            .build();
    }

    @Bean
    public Step processStep(JobRepository jobRepository,
                            PlatformTransactionManager transactionManager) {
        return new StepBuilder("processStep", jobRepository)
            .<Invoice, ProcessedInvoice>chunk(200, transactionManager)
            .reader(reader())
            .processor(processor())
            .writer(writer())
            .build();
    }
}