Устойчивое выполнение процессов в виде Java библиотеки

// Kafka-листенер: дубликатная доставка не должна
// запустить второе задание - id события ключ идемпотентности
@KafkaListener(topics = "orders")
public void onOrderCreated(
    String eventId, String orderId, String userId) {
    try {
        jobSystem.startJob(
            JobOptions.jobWithExternalId(eventId),
            this::processOrder, orderId, userId);
    } catch (JobAlreadyExistsException e) {
        // дубликат: задание уже существует
        log.info("Event {} already has a job", eventId);
    }
}

public void processOrder(String orderId, String userId) {
    // повторяется до успеха - события не теряются
}

// Kafka-листенер: дубликатная доставка не должна
// запустить второе задание - id события ключ идемпотентности
@KafkaListener(topics = "orders")
fun onOrderCreated(
    eventId: String, orderId: String, userId: String
) {
    try {
        jobSystem.startJob(
            JobOptions.jobWithExternalId(eventId),
            this::processOrder, orderId, userId)
    } catch (e: JobAlreadyExistsException) {
        // дубликат: задание уже существует
        log.info("Event {} already has a job", eventId)
    }
}

fun processOrder(orderId: String, userId: String) {
    // повторяется до успеха - события не теряются
}

// Задание получает команды во время выполнения
public void exportReport(ExportRequest request) {
    Signal<ExportCommand> command = Signal.of();
    for (String page : request.pages()) {
        ExportCommand cmd = command.waitAndPop();
        render(page, cmd);
    }
}

// Из любого места - например, из REST-контроллера:
jobSystem.signal(job, new ExportCommand("change-scope"));

// Задание получает команды во время выполнения
fun exportReport(request: ExportRequest) {
    val command: Signal<ExportCommand> = Signal.of()
    for (page in request.pages()) {
        val cmd = command.waitAndPop()
        render(page, cmd)
    }
}

// Из любого места - например, из REST-контроллера:
jobSystem.signal(job, ExportCommand("change-scope"))

@Service
public class ReportScheduler {
    private final RecurringJobScheduler recurring;

    public void register() {
        // каждую ночь в 02:00 - само задание устойчиво
        recurring.scheduleAtCron(
            "nightly-report", "0 0 2 * * *",
            this::generateReport);
    }

    public void generateReport() {
        // автоматически восстанавливается после сбоев
    }
}

@Service
class ReportScheduler(
    private val recurring: RecurringJobScheduler,
) {
    fun register() {
        // каждую ночь в 02:00 - само задание устойчиво
        recurring.scheduleAtCron(
            "nightly-report", "0 0 2 * * *",
            this::generateReport)
    }

    fun generateReport() {
        // автоматически восстанавливается после сбоев
    }
}

// Чанки из БД: сохранённый индекс позволяет продолжить
// после перезапуска без повторной обработки готовых чанков
public void processBacklog() {
    // возвращает сохранённый индекс или 0 при первом запуске
    Integer index = Job.computeState("index",
        Integer.class, current ->
            current == null ? 0 : current);

    // запрашиваем чанки начиная с сохранённого индекса
    List<Chunk> chunks =
        chunkRepository.findFrom(index);

    for (int i = 0; i < chunks.size(); i++) {
        Job.throwIfInterruptedOrCancelled();
        chunkService.handle(chunks.get(i));
        Job.setState("index", index + i + 1);
    }
}

// Чанки из БД: сохранённый индекс позволяет продолжить
// после перезапуска без повторной обработки готовых чанков
fun processBacklog() {
    // возвращает сохранённый индекс или 0 при первом запуске
    val index = Job.computeState("index",
        Integer::class.java) { it ?: 0 }

    // запрашиваем чанки начиная с сохранённого индекса
    val chunks =
        chunkRepository.findFrom(index)

    chunks.forEachIndexed { i, chunk ->
        Job.throwIfInterruptedOrCancelled()
        chunkService.handle(chunk)
        Job.setState("index", index + i + 1)
    }
}

// Saga: каждый шаг регистрирует компенсацию.
// При сбое завершённые шаги откатываются в обратном порядке.
public String placeOrder(String orderId) {
    Deque<Runnable> compensate = new ArrayDeque<>();
    try {
        String paymentId = jobSystem.execute(
            paymentService::charge, String.class, orderId);
        compensate.push(() -> jobSystem.execute(
            paymentService::refund, orderId, paymentId));

        String shipmentId = jobSystem.execute(
            deliveryService::ship, String.class, orderId);
        compensate.push(() -> jobSystem.execute(
            deliveryService::cancelShipment, shipmentId));

        jobSystem.execute(loyaltyService::addPoints, orderId);
        return paymentId;
    } catch (JobFailure e) {
        compensate.forEach(Runnable::run); // обратный порядок
        throw e;
    }
}

// Saga: каждый шаг регистрирует компенсацию.
// При сбое завершённые шаги откатываются в обратном порядке.
fun placeOrder(orderId: String): String {
    val compensate = ArrayDeque<() -> Unit>()
    try {
        val paymentId = jobSystem.execute(
            paymentService::charge, String::class.java, orderId)
        compensate.addFirst {
            jobSystem.execute(
                paymentService::refund, orderId, paymentId)
        }

        val shipmentId = jobSystem.execute(
            deliveryService::ship, String::class.java, orderId)
        compensate.addFirst {
            jobSystem.execute(
                deliveryService::cancelShipment, shipmentId)
        }

        jobSystem.execute(loyaltyService::addPoints, orderId)
        return paymentId
    } catch (e: JobFailure) {
        compensate.forEach { it() } // обратный порядок
        throw e
    }
}

// Задание проверяет отмену между шагами
public void exportReport(ExportRequest request) {
    for (String page : request.pages()) {
        Job.throwIfInterruptedOrCancelled();
        render(page);
    }
}

// Из любого места: отмена каскадно передаётся дочерним заданиям
jobSystem.cancel(job);

// Задание проверяет отмену между шагами
fun exportReport(request: ExportRequest) {
    for (page in request.pages()) {
        Job.throwIfInterruptedOrCancelled()
        render(page)
    }
}

// Из любого места: отмена каскадно передаётся дочерним заданиям
jobSystem.cancel(job)