Spring Boot — Batch Jobs (@Scheduled, chunking, idempotence, reprise)

@Slf4j
@Service
@RequiredArgsConstructor
public class MoveAllStocksJob {

    private final AtomicBoolean running = new AtomicBoolean(false);

    @Scheduled(cron = "0 */1 * * * *")
    public void run() {
        if (!running.compareAndSet(false, true)) { // ✓ atomique — pas de race condition
            log.info("Job skipped: already running");
            return;
        }
        try {
            // traitement...
        } finally {
            running.set(false); // ✓ libération garantie même en cas d'exception
        }
    }
}

@Service
@RequiredArgsConstructor
public class StockChunkProcessor {

    private final EntityManager em;
    private final StockRepository stockRepository;

    @Transactional // ✓ une transaction par chunk, pas pour tout le job
    public Long processNextChunk(String locationId, long lastId, int chunkSize) {
        List<StockPo> chunk = stockRepository.fetchChunk(
            locationId, lastId, PageRequest.of(0, chunkSize)
        );

        if (chunk.isEmpty()) return null;

        for (StockPo stock : chunk) {
            stock.setStatus(MOVED);                          // ✓ écriture idempotente
            stock.setLastUpdatedTime(OffsetDateTime.now());
        }

        em.flush();  // ✓ force le SQL avant de vider le contexte
        em.clear();  // ✓ libère la mémoire Hibernate entre chunks

        return chunk.get(chunk.size() - 1).getId();
    }
}

// ❌ OFFSET — coût croissant, instable si des lignes sont insérées entre deux pages
SELECT * FROM stock OFFSET 10000 LIMIT 100;

// ✓ KEYSET — coût constant, stable, idéal pour les batch jobs
SELECT * FROM stock
WHERE location_id = :locationId
AND id > :lastId
ORDER BY id ASC
LIMIT 100;

@Query("""
    select s from StockPo s
    where s.locationId = :locationId
    and s.id > :lastId
    order by s.id asc
    """)
List<StockPo> fetchChunk(String locationId, Long lastId, Pageable pageable);

// ❌ Pas idempotent — exécuté 2 fois = quantité doublée
stock.setQuantity(stock.getQuantity() + delta);

// ✓ Idempotent — exécuté N fois = même résultat
stock.setStatus(MOVED);

// ✓ Idempotent via vérification préalable
if (stock.getStatus() != MOVED) {
    process(stock);
}

@Entity
@Table(name = "job_checkpoint")
public class JobCheckpoint {
    @Id
    private String jobName;
    private Long lastId = 0L;
    private OffsetDateTime updatedAt;
}

@Scheduled(cron = "0 */1 * * * *")
public void run() {
    if (!running.compareAndSet(false, true)) return;

    try {
        String jobName = "MOVE_ALL_STOCKS:LILLE-01";

        JobCheckpoint checkpoint = checkpointRepository
            .findById(jobName)
            .orElse(new JobCheckpoint(jobName, 0L));

        long lastId = checkpoint.getLastId();

        while (true) {
            Long newLastId = chunkProcessor.processNextChunk(
                "LILLE-01", lastId, CHUNK_SIZE
            );

            if (newLastId == null) break; // ✓ plus de données à traiter

            checkpoint.setLastId(newLastId);     // ✓ checkpoint après chaque chunk
            checkpoint.setUpdatedAt(OffsetDateTime.now());
            checkpointRepository.save(checkpoint);

            lastId = newLastId;
        }
    } catch (Exception e) {
        log.error("Job failed", e);
    } finally {
        running.set(false);
    }
}