KafkaBus 1.0.0

KafkaBus

A lightweight, middleware-enabled Kafka library for .NET that simplifies producer and consumer implementations with a clean, fluent API.

Features

• 🚀 Simple and intuitive API
• 🔌 Built-in dependency injection support
• 🔗 Middleware pipeline for producers and consumers
• ⚡ Batch message production
• 🔄 Manual acknowledgment support
• 🎯 Custom serialization/deserialization
• 📦 Convention-based configuration scanning

Installation

dotnet add package KafkaBus

Quick Start

1. Configure Services

var builder = WebApplication.CreateBuilder(args);

builder.Services.AddKafkaBusProducers(builder.Configuration, [typeof(Program).Assembly]);
builder.Services.AddKafkaBusConsumers(builder.Configuration, [typeof(Program).Assembly]);

var app = builder.Build();
app.Run();

2. Add Configuration

{
  "KafkaBus": {
    "DefaultProducerSettings": {
      "BootstrapServers": "localhost:9092"
    },
    "DefaultConsumerSettings": {
      "BootstrapServers": "localhost:9092",
      "AutoOffsetReset": "Earliest",
      "EnableAutoCommit": true
    }
  }
}

3. Define Your Message

public record OrderCreated(Guid Id, string ProductName, decimal Amount);

4. Send Messages

public class OrderController(IMessageBus messageBus) : ControllerBase
{
    [HttpPost]
    public async Task<IActionResult> CreateOrder(CreateOrderRequest request)
    {
        var message = new OrderCreated(Guid.NewGuid(), request.ProductName, request.Amount);
        
        await messageBus.SendAsync("orders", message);
        
        return Ok();
    }
}

5. Consume Messages

public class OrderCreatedConsumer(ILogger<OrderCreatedConsumer> logger) 
    : MessageConsumer<OrderCreated>
{
    public override Task HandleAsync(ConsumeContext<string, OrderCreated> context, CancellationToken ct)
    {
        logger.LogInformation("Order received: {OrderId}", context.Message.Id);
        return Task.CompletedTask;
    }
}

public class OrderCreatedConsumerConfiguration(IServiceProvider sp) 
    : ConsumerConfiguration<string, OrderCreated>(sp)
{
    public override string Topic => "orders";
}

That's it! Your consumer will automatically start listening to the orders topic.

Producer Features

Basic Usage

// Send to topic (string key)
var result = await messageBus.SendAsync("orders", new OrderCreated(...));

// Send to specific partition
var result = await messageBus.SendAsync("orders", partition: 2, new OrderCreated(...));

// Send batch
var orders = new[] { order1, order2, order3 };
var results = await messageBus.SendBatchAsync("orders", orders);

Delivery Result

The SendAsync method returns a DeliveryResult containing metadata about the produced message:

var result = await messageBus.SendAsync("orders", message);

Console.WriteLine($"Topic: {result.Topic}");
Console.WriteLine($"Partition: {result.Partition.Value}");
Console.WriteLine($"Offset: {result.Offset.Value}");
Console.WriteLine($"Timestamp: {result.Timestamp.UtcDateTime}");
Console.WriteLine($"Key: {result.Key}");
Console.WriteLine($"Status: {result.Status}"); // Persisted, PossiblyPersisted, NotPersisted

// Access the produced message
Console.WriteLine($"Message Value: {result.Message.Value}");
Console.WriteLine($"Message Headers: {result.Message.Headers}");

Batch Result

var results = await messageBus.SendBatchAsync("orders", orders);

foreach (var result in results)
{
    if (result.Status == PersistenceStatus.Persisted)
        Console.WriteLine($"Delivered to P:{result.Partition.Value} O:{result.Offset.Value}");
    else
        Console.WriteLine($"Delivery uncertain: {result.Status}");
}

Custom Key Types

// Send with Guid key
await messageBus.SendAsync<Guid, OrderCreated>("orders", message);

// Send with int key
await messageBus.SendAsync<int, UserEvent>("users", message);

// Send with custom key to specific partition
await messageBus.SendAsync<Guid, OrderCreated>("orders", partition: 1, message);

// Batch with custom key
await messageBus.SendBatchAsync<Guid, OrderCreated>("orders", messages);

Custom Headers

var headers = new Headers
{
    { "correlation-id", Encoding.UTF8.GetBytes(Guid.NewGuid().ToString()) },
    { "source", Encoding.UTF8.GetBytes("order-service") }
};

// With string key
await messageBus.SendAsync("orders", message, headers);

// With custom key
await messageBus.SendAsync<Guid, OrderCreated>("orders", message, headers);

Custom Producer Configuration

Create a custom configuration to control keys, serializers, or Kafka settings:

public class OrderProducerConfiguration(IServiceProvider sp) 
    : ProducerConfiguration<string, OrderCreated>(sp)
{
    // Use OrderId as the message key for partitioning
    public override string GetKey(OrderCreated message) => message.Id.ToString();
    
    // Custom Kafka settings
    public override Task<ProducerConfig> ConfigureAsync()
    {
        var config = defaultConfiguration;
        config.Acks = Acks.All;
        config.EnableIdempotence = true;
        config.MaxInFlight = 5;
        return Task.FromResult(config);
    }
}

Custom Key Configuration with Serializer

public class OrderProducerConfiguration(IServiceProvider sp) 
    : ProducerConfiguration<Guid, OrderCreated>(sp)
{
    public override Guid GetKey(OrderCreated message) => message.Id;
    
    public override ISerializer<Guid>? KeySerializer => new GuidSerializer();
}

Consumer Features

Basic Consumer

public class OrderConsumer : MessageConsumer<OrderCreated>
{
    public override Task HandleAsync(ConsumeContext<string, OrderCreated> context, CancellationToken ct)
    {
        Console.WriteLine($"Received: {context.Message.Id}");
        return Task.CompletedTask;
    }
}

Custom Consumer Configuration

public class OrderConsumerConfiguration(IServiceProvider sp) 
    : ConsumerConfiguration<string, OrderCreated>(sp)
{
    public override string Topic => "orders";           // Custom topic name
    public override string GroupId => "order-service";  // Custom consumer group
    public override int WorkerCount => 4;               // Parallel workers
}

Manual Acknowledgment

For scenarios requiring manual message acknowledgment, disable auto commit in appsettings:

{
  "KafkaBus": {
    "DefaultConsumerSettings": {
      "BootstrapServers": "localhost:9092",
      "EnableAutoCommit": false
    }
  }
}

public class OrderConsumer(ILogger<OrderConsumer> logger) 
    : MessageConsumer<OrderCreated>
{
    public override async Task HandleAsync(ConsumeContext<string, OrderCreated> context, CancellationToken ct)
    {
        try
        {
            await ProcessOrderAsync(context.Message);
            Ack(context); // Manually acknowledge
        }
        catch (Exception ex)
        {
            logger.LogError(ex, "Processing failed, message will be redelivered");
            throw;
        }
    }
}

Batch Acknowledgment

public class BatchOrderConsumer : MessageConsumer<OrderCreated>
{
    private readonly List<ConsumeContext<string, OrderCreated>> _batch = [];
    private const int BatchSize = 100;

    public override async Task HandleAsync(ConsumeContext<string, OrderCreated> context, CancellationToken ct)
    {
        _batch.Add(context);

        if (_batch.Count >= BatchSize)
        {
            await ProcessBatchAsync(_batch.Select(c => c.Message));
            Ack(_batch); // Acknowledge all at once
            _batch.Clear();
        }
    }
}

Accessing Message Metadata

public override Task HandleAsync(ConsumeContext<string, OrderCreated> context, CancellationToken ct)
{
    Console.WriteLine($"Topic: {context.Topic}");
    Console.WriteLine($"Partition: {context.Partition}");
    Console.WriteLine($"Offset: {context.Offset}");
    Console.WriteLine($"Timestamp: {context.Timestamp}");
    Console.WriteLine($"Key: {context.Key}");
    
    // Access headers
    if (context.Headers?.TryGetLastBytes("correlation-id", out var bytes) == true)
    {
        var correlationId = Encoding.UTF8.GetString(bytes);
    }
    
    return Task.CompletedTask;
}

Custom Key Consumer

public class OrderConsumer : MessageConsumer<Guid, OrderCreated>
{
    public override Task HandleAsync(ConsumeContext<Guid, OrderCreated> context, CancellationToken ct)
    {
        Console.WriteLine($"Key: {context.Key}, Order: {context.Message.Id}");
        return Task.CompletedTask;
    }
}

public class OrderConsumerConfiguration(IServiceProvider sp) 
    : ConsumerConfiguration<Guid, OrderCreated>(sp)
{
    public override string Topic => "orders";
    public override IDeserializer<Guid>? KeyDeserializer => new GuidDeserializer();
}

Middleware

Middlewares allow you to intercept and process messages in a pipeline pattern.

Producer Middleware

[MiddlewareOrder(1)]
public class ProducerLoggingMiddleware<TKey, TMessage>(ILogger<ProducerLoggingMiddleware<TKey, TMessage>> logger) 
    : IProduceMiddleware<TKey, TMessage>
{
    public async Task<DeliveryResult<TKey, TMessage>> InvokeAsync(
        ProduceContext<TKey, TMessage> context,
        ProduceDelegate<TKey, TMessage> next,
        CancellationToken ct)
    {
        var sw = Stopwatch.StartNew();
        logger.LogInformation("Producing to {Topic}", context.Topic);

        var result = await next(context, ct);

        logger.LogInformation("Produced to {Topic} P:{Partition} O:{Offset} in {Elapsed}ms",
            context.Topic, result.Partition.Value, result.Offset.Value, sw.ElapsedMilliseconds);

        return result;
    }
}

Consumer Middleware

[MiddlewareOrder(1)]
public class ConsumerLoggingMiddleware<TKey, TMessage>(ILogger<ConsumerLoggingMiddleware<TKey, TMessage>> logger) 
    : IConsumeMiddleware<TKey, TMessage>
{
    public async Task InvokeAsync(
        ConsumeContext<TKey, TMessage> context,
        ConsumeDelegate<TKey, TMessage> next,
        CancellationToken ct)
    {
        var sw = Stopwatch.StartNew();
        logger.LogInformation("Consuming from {Topic} P:{Partition} O:{Offset}",
            context.Topic, context.Partition, context.Offset);

        await next(context, ct);

        logger.LogInformation("Consumed in {Elapsed}ms", sw.ElapsedMilliseconds);
    }
}

Middleware Order

Use [MiddlewareOrder] attribute to control execution order. Higher values execute first (outermost in pipeline).

[MiddlewareOrder(100)] // Executes first
public class AuthenticationMiddleware<TKey, TMessage> : IConsumeMiddleware<TKey, TMessage> { }

[MiddlewareOrder(50)]  // Executes second
public class ValidationMiddleware<TKey, TMessage> : IConsumeMiddleware<TKey, TMessage> { }

[MiddlewareOrder(1)]   // Executes last (closest to handler)
public class LoggingMiddleware<TKey, TMessage> : IConsumeMiddleware<TKey, TMessage> { }

Custom Serialization

Custom Serializer

public class ProtobufSerializer<T> : ISerializer<T> where T : IMessage<T>
{
    public byte[] Serialize(T data, SerializationContext context)
    {
        return data.ToByteArray();
    }
}

Custom Deserializer

public class ProtobufDeserializer<T> : IDeserializer<T> where T : IMessage<T>, new()
{
    private readonly MessageParser<T> _parser = new(() => new T());

    public T Deserialize(ReadOnlySpan<byte> data, bool isNull, SerializationContext context)
    {
        if (isNull || data.IsEmpty) return default!;
        return _parser.ParseFrom(data.ToArray());
    }
}

Register Default Serializers

builder.Services.AddKafkaBusProducers(
    builder.Configuration,
    [typeof(Program).Assembly],
    defaultValueSerializer: typeof(ProtobufSerializer<>));

builder.Services.AddKafkaBusConsumers(
    builder.Configuration,
    [typeof(Program).Assembly],
    defaultValueDeserializer: typeof(ProtobufDeserializer<>));

Advanced Configuration

Custom Configuration Section Names

builder.Services.AddKafkaBusProducers(
    builder.Configuration,
    [typeof(Program).Assembly],
    sectionName: "Messaging:KafkaProducer");

builder.Services.AddKafkaBusConsumers(
    builder.Configuration,
    [typeof(Program).Assembly],
    sectionName: "Messaging:KafkaConsumer");

{
  "Messaging": {
    "KafkaProducer": {
      "BootstrapServers": "kafka:9092",
      "Acks": "All"
    },
    "KafkaConsumer": {
      "BootstrapServers": "kafka:9092",
      "AutoOffsetReset": "Earliest",
      "EnableAutoCommit": true
    }
  }
}

Multiple Assemblies

builder.Services.AddKafkaBusProducers(builder.Configuration, [
    typeof(Program).Assembly,
    typeof(OrderCreated).Assembly,
    typeof(SharedMessages).Assembly
]);

API Reference

IMessageBus

ConsumeContext<TKey, TMessage>

MessageConsumer<TMessage>

License

This project is licensed under the MIT License - see the LICENSE file for details.