Error Handling

Pipelite offers two error handling modes at the flow level: the retry channel and the custom error channel. Both are configured via the DSL at the end of the flow definition.

Retry Channel

The retry channel is the simplest mode. When an exception propagates through the flow, the framework serializes the Exchange into a FlowExecutionDump and re-queues it for a subsequent attempt.

FlowDefinition flow = Pipelite.defineFlow("resilient-flow")
    .fromSource("kafka://my-topic")
    .process("parse", exchange -> {
        // may throw exceptions
    })
    .toSink("slf4j://output")
    .withRetryChannel()    (1)
    .build();
1 Enables automatic retry with the default strategy.

Internal mechanism

  1. The exception is caught by RetryChannelExceptionHandler.

  2. The Exchange is serialized into a FlowExecutionDump via FlowExecutionDumpFactory.

  3. The dump is persisted in FlowExecutionDumpRepository (default in-memory implementation: FlowExecutionDumpInMemoryRepository).

  4. The RetryPollingConsumer periodically retrieves the dumps and retries execution.

  5. RetryStrategyFilter determines whether the attempt should be made based on the configured strategy.

Custom Error Channel

For more flexible error handling, use withErrorChannel:

FlowDefinition flow = Pipelite.defineFlow("flow-with-error-channel")
    .fromSource("http://api/v1/ingest")
    .process("process", exchange -> { /* ... */ })
    .toSink("slf4j://output")
    .withErrorChannel(err -> err.toChannel("link://ingest-dead-letter"))   (1)
    .build();
1 The target is a URL, like every destination in the DSL: link://<source endpoint name> for an internal flow (the name its fromSource(…​) declares, 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.

Default handler

If no error channel 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.

Class Description

FlowExecutionDump (interface)

Contract for the execution dump

DefaultFlowExecutionDump

Standard in-memory implementation

SerializedFlowExecutionDump

Base64-serialized version for persistence

FlowExecutionDumpInMemoryRepository

In-memory repository (not persistent across restarts)
For production environments with durability requirements, FlowExecutionDumpRepository can be implemented with a persistent backend (e.g. database, Redis).