PollyKafka 1.0.0

PollyKafka

Polly v8 resilience for Confluent.Kafka — automatic retry, circuit breaker, and per-operation timeout for Kafka producers and consumers. Drop-in wrappers, no custom serialisers required.

Why PollyKafka?

Kafka brokers fail transiently: leader elections, broker restarts, network blips. Without resilience, a single BrokerNotAvailable or RequestTimedOut error can crash your producer or leave your consumer in a broken state. PollyKafka wraps every produce and consume operation in a Polly v8 pipeline so transient failures are retried automatically, persistent broker failures trip the circuit breaker, and hanging operations are cancelled by a configurable timeout.

Installation

dotnet add package PollyKafka

Quick Start

Producer

// Manual construction
var producerConfig = new ProducerConfig { BootstrapServers = "localhost:9092" };
var inner = new ProducerBuilder<string, string>(producerConfig).Build();
var producer = new ResilientProducer<string, string>(inner, new PollyKafkaOptions
{
    MaxRetries  = 3,
    BaseDelay   = TimeSpan.FromMilliseconds(200),
});

var result = await producer.ProduceAsync("my-topic", new Message<string, string>
{
    Key   = "order-id",
    Value = System.Text.Json.JsonSerializer.Serialize(order),
});
Console.WriteLine($"Delivered to {result.TopicPartitionOffset}");

Consumer

var consumerConfig = new ConsumerConfig
{
    BootstrapServers = "localhost:9092",
    GroupId          = "my-consumer-group",
    AutoOffsetReset  = AutoOffsetReset.Earliest,
};
var inner    = new ConsumerBuilder<string, string>(consumerConfig).Build();
var consumer = new ResilientConsumer<string, string>(inner, new PollyKafkaOptions());

consumer.Subscribe("my-topic");

while (!stoppingToken.IsCancellationRequested)
{
    var result = await consumer.ConsumeAsync(stoppingToken);
    if (result is null) continue; // timeout — no message available

    // process result.Message.Value ...
    consumer.Commit(result);
}

With Dependency Injection

// Program.cs
builder.Services.AddResilientKafkaProducer<string, string>(
    new ProducerConfig { BootstrapServers = "localhost:9092" },
    o =>
    {
        o.MaxRetries      = 3;
        o.OperationTimeout = TimeSpan.FromSeconds(15);
    });

builder.Services.AddResilientKafkaConsumer<string, string>(
    new ConsumerConfig
    {
        BootstrapServers = "localhost:9092",
        GroupId          = "orders-consumer",
    });

// Inject ResilientProducer<string, string> / ResilientConsumer<string, string>

Configuration

var options = new PollyKafkaOptions
{
    // Retry
    MaxRetries = 3,                              // 0 = no retry
    BaseDelay  = TimeSpan.FromMilliseconds(200), // exponential base
    MaxDelay   = TimeSpan.FromSeconds(30),

    // Circuit breaker
    CircuitBreakerFailureRatio      = 0.5,
    CircuitBreakerMinimumThroughput = 10,
    CircuitBreakerSamplingDuration  = TimeSpan.FromSeconds(30),
    CircuitBreakerBreakDuration     = TimeSpan.FromSeconds(5),

    // Timeout
    OperationTimeout = TimeSpan.FromSeconds(10),

    // Which Kafka error codes trigger retry/CB
    TransientErrorCodes = new HashSet<ErrorCode>
    {
        ErrorCode.BrokerNotAvailable,
        ErrorCode.LeaderNotAvailable,
        ErrorCode.NotLeaderForPartition,
        ErrorCode.RequestTimedOut,
        ErrorCode.NetworkException,
        ErrorCode.KafkaStorageError,
        ErrorCode.Local_AllBrokersDown,
        ErrorCode.Local_TimedOut,
        ErrorCode.Local_Transport,
        ErrorCode.Local_MsgTimedOut,
    },
};

Error Handling

Non-transient KafkaExceptions (e.g. TopicAuthorizationFailed) are rethrown as-is. After retries are exhausted the last exception is a TransientKafkaException wrapping the original:

try
{
    await producer.ProduceAsync("my-topic", message);
}
catch (TransientKafkaException ex)
{
    // All retries failed
    Console.WriteLine($"Kafka error: {ex.ErrorCode} — {ex.KafkaException.Message}");
}
catch (BrokenCircuitException)
{
    // Circuit is open — fail fast
}
catch (TimeoutRejectedException)
{
    // Operation exceeded OperationTimeout
}

Resilience Pipeline Order

Operations flow through the pipeline in this order:

Retry → Circuit Breaker → Timeout → Kafka operation

Related Packages

License

MIT