KafkaIO 1.0.26

KafkaIO - A Fluent Kafka Configuration Library for .NET

Overview

KafkaIO is a .NET library that provides a fluent interface for configuring Kafka producers and consumers with Microsoft's dependency injection system. It simplifies Kafka client setup with built-in support for common patterns like circuit breakers, retries, batch processing, and outbox patterns.

Features

• Fluent Configuration API - Easy-to-use builder pattern for setting up producers and consumers
• Dependency Injection Integration - Seamless integration with IServiceCollection
• Built-in Resilience Patterns:
  • Circuit breakers
  • Retry strategies
  • Batch processing
  • Outbox pattern
• Serialization Support - JSON serialization out of the box
• Producer Features:
  • Message batching
  • Error handling
  • Configuration options for all Confluent.Kafka producer settings
• Consumer Features:
  • Topic subscription
  • Message handling
  • Configuration options for all Confluent.Kafka consumer settings

Installation

dotnet add package KafkaIO

Quick Start

Basic Producer/Consumer Setup

using KafkaIO;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;

var builder = Host.CreateApplicationBuilder();

builder.Services.AddKafkaIO(kafka => kafka
    .AddBootstrapServers("localhost:9092")
    // Configure producer
    .AddProducer("sample.topic")
    // Configure consumer
    .AddConsumer("sample.group", consumer => consumer
        .AddTopic("sample.topic"));

var app = builder.Build();
await app.StartAsync();

// Get producer and send message
var producer = app.Services.GetKeyedService<IProducer>("sample.topic")!;
await producer.ProduceAsync(new SampleMessage { Text = "Hello Kafka!" });

await app.StopAsync();

Advanced Configuration

builder.Services.AddKafkaIO(kafka => kafka
    .AddBootstrapServers("localhost:9092")
    .AddProducer("orders", producer => producer
        .AddSerializer(s => s.UseJsonSerializer())
        .AddCircuitBreaker(cb => cb
            .WithFailureThreshold(5)
            .WithOpenStateDuration(TimeSpan.FromSeconds(30)))
        .AddBatchProducer(batch => batch
            .WithMessageLimit(100)
            .WithTimeLimit(1000))
        .AddMessageRetry(retry => retry
            .WithIncremental(
                retryLimit: 3,
                initialInterval: TimeSpan.FromSeconds(1),
                intervalIncrement: TimeSpan.FromSeconds(2)))
    .AddConsumer("order-processors", consumer => consumer
        .AddTopic("orders")
        .AddDeserializer(d => d.UseJsonSerializer())
        .WithAutoOffsetReset(AutoOffsetReset.Earliest)
        .HandleType<Order>(h => h
            .AddHandler<OrderProcessor>())));

Configuration Options

Producer Configuration

Consumer Configuration

Resilience Patterns

Circuit Breaker

.AddCircuitBreaker(cb => cb
    .WithFailureThreshold(5)
    .WithOpenStateDuration(TimeSpan.FromSeconds(30))

Retry Strategies

// Fixed intervals
.AddMessageRetry(retry => retry
    .WithIntervals(1000, 2000, 3000))

// Incremental intervals
.AddMessageRetry(retry => retry
    .WithIncremental(
        retryLimit: 3,
        initialInterval: TimeSpan.FromSeconds(1),
        intervalIncrement: TimeSpan.FromSeconds(2)))

// Immediate retry
.AddMessageRetry(retry => retry
    .WithImmediate(retryLimit: 5))

Outbox Pattern

.AddOutbox(outbox => outbox
    .UseInMemory()
    .WithPollingPeriod(2000))

Message Handling

Define message handlers by implementing IConsumerHandler<T>:

public class OrderProcessor : IConsumerHandler<Order>
{
    public Task HandleAsync(ConsumeResult<string, Order> message, CancellationToken cancellationToken)
    {
        // Process order
        Console.WriteLine($"Processing order {message.Value.OrderId}");
        return Task.CompletedTask;
    }
}

Advanced Topics

Custom Serialization

Implement ISerializer/IDeserializer interfaces for custom serialization:

public class ProtobufSerializer : ISerializer
{
    public byte[] Serialize<T>(T data) => /* serialize */;
}

License

MIT License

Contributing

Contributions are welcome! Please open an issue or submit a pull request.