[Feature]: Unified Error Handling for Messaging Integrations

[eneshoxha]

Summary

This issue documents the implementation of unified, stream-level error handling across all messaging integration sink operators, replacing the previous per-operator error handling approach.

Problem Statement

Previously, each messaging integration (Kafka, Pulsar, RabbitMQ, AWS SQS, Azure Service Bus) implemented its own error handling with custom parameters:

// OLD: Each operator had its own error-handling parameters
new KafkaSinkOperator<Order>(
    bootstrapServers: "localhost:9092",
    topic: "orders",
    maxRetries: 3,                              // ❌ Duplicated across integrations
    retryDelayMs: 100,                          // ❌ Inconsistent behavior
    errorHandler: (ex, msg) => { ... }          // ❌ Per-operator configuration
);

Issues with the Previous Approach

1. Code Duplication: Each integration had its own retry/error logic
2. Inconsistent Behavior: Different integrations might handle errors differently
3. Configuration Complexity: Error handling configured per-operator, not centrally
4. No Integration with Core: Didn't leverage the existing StreamExecutionOptions infrastructure

Solution

1. Core Library Changes

Made the error handling infrastructure public so external integrations can use it:

Cortex.Streams/ErrorHandling/ErrorHandlingHelper.cs

// Changed from internal to public
public static class ErrorHandlingHelper
{
    public static bool TryExecute<TInput>(
        StreamExecutionOptions options,
        string operatorName,
        object rawInput,
        Action<TInput> action) { ... }
}

Cortex.Streams/ErrorHandling/StreamExecutionOptions.cs

// Made Default public
public static readonly StreamExecutionOptions Default = new StreamExecutionOptions();

2. Integration Sink Operators

All sink operators now implement IErrorHandlingEnabled:

Operator	Package
KafkaSinkOperator<TInput>	Cortex.Streams.Kafka
KafkaSinkOperator<TKey, TValue>	Cortex.Streams.Kafka
PulsarSinkOperator<TInput>	Cortex.Streams.Pulsar
RabbitMQSinkOperator<TInput>	Cortex.Streams.RabbitMQ
SQSSinkOperator<TInput>	Cortex.Streams.AWSSQS
AzureServiceBusSinkOperator<TInput>	Cortex.Streams.AzureServiceBus

New Pattern:

public class KafkaSinkOperator<TInput> : ISinkOperator<TInput>, IErrorHandlingEnabled, IDisposable
{
    private static readonly string OperatorName = $"KafkaSinkOperator<{typeof(TInput).Name}>";
    private StreamExecutionOptions _executionOptions = StreamExecutionOptions.Default;

    public void SetErrorHandling(StreamExecutionOptions options)
    {
        _executionOptions = options ?? StreamExecutionOptions.Default;
    }

    public void Process(TInput input)
    {
        // Use core error handling
        ErrorHandlingHelper.TryExecute(
            _executionOptions,
            OperatorName,
            input,
            (Action<TInput>)ProduceMessage);
    }
}

3. Operator Adapters & FanOut Support

Fixed critical bug where StreamExecutionOptions were not being forwarded to integration sink operators:

SinkOperatorAdapter<T> - Now implements IErrorHandlingEnabled and forwards to wrapped operator:

public void SetErrorHandling(StreamExecutionOptions options)
{
    if (_sinkOperator is IErrorHandlingEnabled errorHandlingEnabled)
    {
        errorHandlingEnabled.SetErrorHandling(options);
    }
}

BranchOperator<T> and ForkOperator<T> - Now forward error handling to their inner operators for FanOut support.

Usage

Simple Stream with Error Handling

var stream = StreamBuilder<Order, Order>
    .CreateNewStream("order-processor")
    .WithExecutionOptions(new StreamExecutionOptions
    {
        ErrorHandlingStrategy = ErrorHandlingStrategy.Retry,
        MaxRetries = 5,
        RetryDelay = TimeSpan.FromSeconds(1)
    })
    .Stream(sourceOperator)
    .Map(order => ProcessOrder(order))
    .Sink(new KafkaSinkOperator<Order>("localhost:9092", "orders"))
    .Build();

FanOut with Unified Error Handling

var stream = StreamBuilder<Order, Order>
    .CreateNewStream("order-fanout")
    .WithExecutionOptions(new StreamExecutionOptions
    {
        ErrorHandlingStrategy = ErrorHandlingStrategy.Skip,  // Skip failed messages
        OnError = ctx => 
        {
            logger.LogError(ctx.Exception, 
                "Error in {Operator} processing {Input}", 
                ctx.OperatorName, ctx.Input);
            return ErrorHandlingDecision.Skip;
        }
    })
    .Stream(sourceOperator)
    .FanOut()
        .To("kafka", new KafkaSinkOperator<Order>("kafka:9092", "orders"))
        .To("rabbitmq", new RabbitMQSinkOperator<Order>("rabbitmq", "orders"))
        .To("sqs", new SQSSinkOperator<Order>("https://sqs.aws/queue"))
    .Build();

Custom Per-Error Decision

.WithExecutionOptions(new StreamExecutionOptions
{
    OnError = ctx => 
    {
        // Retry transient errors
        if (ctx.Exception is TimeoutException || ctx.Exception is HttpRequestException)
            return ErrorHandlingDecision.Retry;
        
        // Skip serialization errors
        if (ctx.Exception is JsonException)
            return ErrorHandlingDecision.Skip;
        
        // Stop on critical errors
        if (ctx.Exception is AuthenticationException)
            return ErrorHandlingDecision.Stop;
        
        // Default: rethrow
        return ErrorHandlingDecision.Rethrow;
    }
})

Error Handling Flow

WithExecutionOptions(options)
    ↓
StreamBuilder._executionOptions = options
    ↓
Build() → new Stream(..., executionOptions)
    ↓
Stream.InitializeErrorHandling(_operatorChain)
    ↓
Recursively traverses operator chain via IHasNextOperators
    ↓
For each IErrorHandlingEnabled operator:
    ↓
operator.SetErrorHandling(options)
    ↓
SinkOperatorAdapter → forwards to KafkaSinkOperator
ForkOperator → forwards to all BranchOperators
BranchOperator → forwards to inner operators

Breaking Changes

Constructor Parameter Changes

The following parameters have been removed from integration sink operators:

• maxRetries
• retryDelayMs
• errorHandler
• maxQueueSize

Migration:

// OLD
new KafkaSinkOperator<Order>(
    bootstrapServers: "localhost:9092",
    topic: "orders",
    maxRetries: 5,
    retryDelayMs: 1000,
    errorHandler: (ex, msg) => Console.WriteLine(ex)
);

// NEW
// Configure at stream level instead:
.WithExecutionOptions(new StreamExecutionOptions
{
    ErrorHandlingStrategy = ErrorHandlingStrategy.Retry,
    MaxRetries = 5,
    RetryDelay = TimeSpan.FromSeconds(1),
    OnError = ctx => { Console.WriteLine(ctx.Exception); return ErrorHandlingDecision.Retry; }
})
.Sink(new KafkaSinkOperator<Order>("localhost:9092", "orders"))

Benefits

Aspect	Before	After
Configuration	Per-operator	Centralized at stream level
Consistency	Different per integration	Unified behavior
Code	Duplicated retry logic	Single ErrorHandlingHelper
Flexibility	Fixed strategy	Dynamic per-error decisions
Observability	Manual logging	Rich StreamErrorContext
FanOut	No support	Full support across branches

Related Issues

• Relates to core error handling infrastructure in Cortex.Streams.ErrorHandling
• Enables consistent error handling across all messaging integrations
• Supports both simple streams and complex FanOut topologies