PollyKafka 1.0.2

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

Package	Downloads	Description
PollyHealthChecks		ASP.NET Core health checks for Polly v8 circuit breakers — expose circuit-breaker state (Closed, HalfOpen, Open, Isolated) as /health endpoint responses
PollyBackoff		Backoff delay strategies for Polly v8 resilience pipelines
PollyGrpc		Polly v8 resilience interceptor for gRPC
PollyEFCore		Polly v8 resilience pipelines for Entity Framework Core — wrap every EF Core query and SaveChanges with retry, timeout and circuit-breaker via a single AddPollyResilience() call
PollyMailKit		Polly v8 resilience pipelines for MailKit — retry, timeout, and circuit-breaker for SmtpClient.SendAsync and any MailKit SMTP operation
PollyMassTransit		Polly v8 resilience pipelines for MassTransit — retry, timeout, and circuit-breaker for IBus.Publish and ISendEndpointProvider.Send
PollyOpenAI		Polly v8 resilience for OpenAI and Azure OpenAI API calls
PollyAzureEventHub		Polly v8 resilience pipelines for Azure Event Hubs — retry, timeout, and circuit-breaker for EventHubProducerClient and EventHubConsumerClient
PollySignalR		Polly v8 reconnect policy for SignalR
PollyElasticsearch		Polly v8 resilience pipelines for Elastic.Clients.Elasticsearch 8+ — retry, timeout, and circuit-breaker for any Elasticsearch operation, plus a built-in ElasticTransientErrors predicate covering rate limiting (429), service unavailability (503), gateway timeouts (504), and connection failures
PollyHangfire		Polly v8 resilience pipelines for Hangfire — retry, timeout, and circuit-breaker for IBackgroundJobClient.Enqueue and Schedule
PollySendGrid		Polly v8 resilience pipelines for SendGrid — retry, timeout, and circuit-breaker for ISendGridClient.SendEmailAsync
PollyMediatR		Polly v8 resilience pipelines for MediatR — add retry, timeout, circuit-breaker, rate-limiting, hedging, and chaos engineering to any MediatR request handler with a single line of DI registration
PollyAzureKeyVault		Polly v8 resilience pipelines for Azure Key Vault — retry, timeout, and circuit-breaker for SecretClient, KeyClient, and CertificateClient
PollyAzureQueueStorage		Polly v8 resilience pipelines for Azure Queue Storage — retry, timeout, and circuit-breaker for Azure.Storage.Queues QueueClient
PollyRedis		Polly v8 resilience for StackExchange.Redis
PollyAzureServiceBus		Polly v8 resilience for Azure Service Bus — retry, circuit breaker, and timeout for sending and receiving messages
PollyAzureTableStorage		Polly v8 resilience pipelines for Azure Table Storage — retry, timeout, and circuit-breaker for Azure.Data.Tables TableClient
PollyCaching		A caching resilience strategy for Polly v8 pipelines
PollyChaos		Chaos engineering and fault-injection resilience strategies for Polly v8 pipelines
PollyBulkhead		Bulkhead isolation strategy for Polly v8 resilience pipelines

💼 Need .NET consulting?

The author of this package is available for consulting on Polly v8 resilience, Azure cloud architecture, and clean .NET design.

→ solidqualitysolutions.com · LinkedIn

License

MIT