This version is still in development and is not considered stable yet.

Error Handling

Pipelite offers three mutually-exclusive error handling modes at the flow level: retry, a custom error channel, and a fully custom exception handler. Exactly one is configured via the DSL at the end of the flow definition - each returns a terminal builder step, so none can be chained with another (a custom ExceptionHandler configured alongside a retry or an error channel is a compile error, not a silently-ignored combination).

Retry

When an exception propagates through the flow, the framework serializes the Exchange into a FlowExecutionDump and re-queues it for a subsequent attempt, resuming at the step that failed rather than replaying the flow from its source.

An exhaustion action is mandatory: the withRetry(…) lambda only type-checks once it ends in onErrorChannel(…) or onExceptionHandler(…) - a retry with no defined outcome once attempts run out cannot be expressed.

import io.pipelite.dsl.definition.builder.Backoff;
import java.time.Duration;

FlowDefinition flow = Pipelite.defineFlow("resilient-flow")
    .fromSource("kafka://my-topic")
    .process("parse", (exchange, contribution) -> {
        // may throw exceptions
    })
    .toSink("slf4j://output")
    .withRetry(retry -> retry
        .maxAttempts(3)                                          (1)
        .backoff(Backoff.exponential(Duration.ofSeconds(2)))     (2)
        .onErrorChannel(err -> err.toDLQ()))                     (3)
    .build();

1	Up to 3 attempts total (the original send plus 2 retries). Optional - with no `maxAttempts`, only the framework’s own retry-channel polling cadence paces attempts.
2	The delay before each attempt, counted from the previous one. `Backoff.linear(…)` grows by a constant step (2s, 4s, 6s, …); `Backoff.exponential(…)` doubles it each time (2s, 4s, 8s, …). Optional - with no backoff declared, a retry runs as soon as the retry channel next polls.
3	The exhaustion action: here, the built-in dead letter queue (see Custom Error Channel). `onExceptionHandler(handler)` invokes a supplied `ExceptionHandler` instead of routing anywhere.

Custom Error Channel

For error handling with no retry attempts - a single hand-off, not a schedule of them - use withErrorChannel, or onErrorChannel(…) as a retry’s own exhaustion action (above):

FlowDefinition flow = Pipelite.defineFlow("flow-with-error-channel")
    .fromSource("http://api/v1/ingest")
    .process("process", (exchange, contribution) -> { /* ... */ })
    .toSink("slf4j://output")
    .withErrorChannel(err -> err.toChannel("queue://ingest-dead-letter"))   (1)
    .build();

1 The target is a URL, like every destination in the DSL: queue://<queue name> for an internal flow (the queue its fromSource("queue://…") reads, not the flow’s own name), or a registered channel adapter’s protocol (kafka://…) for an external system. A bare name is rejected when the flow is defined.

err.toDLQ() targets the framework’s built-in dead letter queue instead, with no flow to write.

Custom Exception Handler

For anything the two options above don’t cover, withExceptionHandler(handler) hands the exception straight to a user-supplied ExceptionHandler, with no dump, no retry, and no routing of any kind - the handler decides everything.

Default handler

If none of the three is configured, the exception is handled by GlobalDefaultExceptionHandler, which logs the error and stops processing without propagating the exception to the caller.

FlowExecutionDump

The FlowExecutionDump is the serialization of the Exchange at the time of failure. It contains the serialized payload, headers, and properties required for the retry.

The default FlowExecutionDumpRepository is already durable - a pending retry is written to disk before the flow moves on, and survives a restart. Swap it via ConfigurablePipeliteContext#setFlowExecutionDumpRepository(…) for a different backend (e.g. a database or Redis) or to opt back into FlowExecutionDumpInMemoryRepository.