Workflow durable execution, as a Java library

// Kafka listener: a duplicated delivery must not
// start a second job - event id is the idempotency key
@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) {
        // duplicated delivery: job already exists
        log.info("Event {} already has a job", eventId);
    }
}

public void processOrder(String orderId, String userId) {
    // retried until success - no lost events
}

// Kafka listener: a duplicated delivery must not
// start a second job - event id is the idempotency key
@KafkaListener(topics = "orders")
fun onOrderCreated(
    eventId: String, orderId: String, userId: String
) {
    try {
        jobSystem.startJob(
            JobOptions.jobWithExternalId(eventId),
            this::processOrder, orderId, userId)
    } catch (e: JobAlreadyExistsException) {
        // duplicated delivery: job already exists
        log.info("Event {} already has a job", eventId)
    }
}

fun processOrder(orderId: String, userId: String) {
    // retried until success - no lost events
}

// The job receives commands while it is running
public void exportReport(ExportRequest request) {
    Signal<ExportCommand> command = Signal.of();
    for (String page : request.pages()) {
        ExportCommand cmd = command.waitAndPop();
        render(page, cmd);
    }
}

// From anywhere - e.g. a REST controller:
jobSystem.signal(job, new ExportCommand("change-scope"));

// The job receives commands while it is running
fun exportReport(request: ExportRequest) {
    val command: Signal<ExportCommand> = Signal.of()
    for (page in request.pages()) {
        val cmd = command.waitAndPop()
        render(page, cmd)
    }
}

// From anywhere - e.g. a REST controller:
jobSystem.signal(job, ExportCommand("change-scope"))

@Service
public class ReportScheduler {
    private final RecurringJobScheduler recurring;

    public void register() {
        // nightly at 02:00 - itself a durable job
        recurring.scheduleAtCron(
            "nightly-report", "0 0 2 * * *",
            this::generateReport);
    }

    public void generateReport() {
        // recovered automatically after crashes
    }
}

@Service
class ReportScheduler(
    private val recurring: RecurringJobScheduler,
) {
    fun register() {
        // nightly at 02:00 - itself a durable job
        recurring.scheduleAtCron(
            "nightly-report", "0 0 2 * * *",
            this::generateReport)
    }

    fun generateReport() {
        // recovered automatically after crashes
    }
}

// Chunks from the DB: the saved index lets us resume
// after a restart without reprocessing finished chunks
public void processBacklog() {
    // returns the saved index or 0 on the first run
    Integer index = Job.computeState("index",
        Integer.class, current ->
            current == null ? 0 : current);

    // fetch chunks starting from the saved index
    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);
    }
}

// Chunks from the DB: the saved index lets us resume
// after a restart without reprocessing finished chunks
fun processBacklog() {
    // returns the saved index or 0 on the first run
    val index = Job.computeState("index",
        Integer::class.java) { it ?: 0 }

    // fetch chunks starting from the saved index
    val chunks =
        chunkRepository.findFrom(index)

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

// Saga: each step registers its compensation.
// On failure, completed steps roll back in reverse order.
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); // reverse order
        throw e;
    }
}

// Saga: each step registers its compensation.
// On failure, completed steps roll back in reverse order.
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() } // reverse order
        throw e
    }
}

// The job checks for cancellation between steps
public void exportReport(ExportRequest request) {
    for (String page : request.pages()) {
        Job.throwIfInterruptedOrCancelled();
        render(page);
    }
}

// From anywhere: cancellation cascades to child jobs
jobSystem.cancel(job);

// The job checks for cancellation between steps
fun exportReport(request: ExportRequest) {
    for (page in request.pages()) {
        Job.throwIfInterruptedOrCancelled()
        render(page)
    }
}

// From anywhere: cancellation cascades to child jobs
jobSystem.cancel(job)