Chd.Library.Kafka 8.7.0

Kafka Helper Library for .NET Core

Chd (Cleverly Handle Difficulty) library helps you cleverly handle difficulty, write code quickly, and keep your application stable.

📝 Table of Contents

• About
• Features
• Installation
• Quick Start
• Configuration
• Producing Messages
• Consuming Messages
• Dependency Injection Patterns
• Advanced Usage
  • Multiple Topics & Consumers
  • Typed Message Consumption
  • Error Handling
• API Reference
• Authors
• Acknowledgements

About

Chd.Library.Kafka is a simple yet powerful .NET Core helper library for Apache Kafka.
It abstracts configuration, connection management and (de)serialization, making it easy to produce/consume messages using familiar DI patterns.

Leverages Confluent.Kafka and Newtonsoft.Json.

Features

• Dependency injection: AddKafkaProducer, AddKafkaConsumer<T>(topic)
• Interface-based abstraction for message production (IMessageProducer)
• Automatic JSON (de)serialization for objects
• Register multiple consumers per topic/service
• Delivery callback and error extensibility for advanced scenarios
• Hosted service consumer: background tasks with DI support
• Configurable via appsettings.json
• .NET 9.0+ support
• Easily testable with Swagger, Web API, or console entry points

Installation

dotnet add package Chd.Library.Kafka

Quick Start

1. Configure Kafka in appsettings.json

"Kafka": {
  "Host": "127.0.0.1",
  "Port": "9092"
}

2. Register Producer and (optionally, multiple) Consumers in DI

var builder = WebApplication.CreateBuilder(args);

builder.Services.AddKafkaProducer();
builder.Services.AddKafkaConsumer<EmailMessageConsumer>("topic1");
builder.Services.AddKafkaConsumer<NotificationConsumer>("topic2");

Producing Messages

Define your message contract:

public class EmailMessage
{
    public string Subject { get; set; }
    public string Content { get; set; }
    public string To { get; set; }
}

Inject the MessageProducer and send messages:

using Library.Kafka;
using Microsoft.AspNetCore.Mvc;

[ApiController]
[Route("kafka")]
public class KafkaTestController : ControllerBase
{
    private readonly MessageProducer _messageProducer;
    public KafkaTestController(MessageProducer messageProducer)
    {
        _messageProducer = messageProducer;
    }

    [HttpGet("AddMessagesToKafkaQueue")]
    public string AddMessagesToKafkaQueue()
    {
        _messageProducer.Produce("topic1", new EmailMessage {
            Subject = "Test1 Subject", Content = "Test1 Content", To = "test1@gmail.com"
        });
        System.Threading.Thread.Sleep(3000);
        _messageProducer.Produce("topic2", new EmailMessage {
            Subject = "Test2 Subject", Content = "Test2 Content", To = "test2@gmail.com"
        });
        System.Threading.Thread.Sleep(3000);
        return "Ok";
    }
}

• Any object will be serialized to JSON automatically.
• Producer is disposable; managed lifetime by DI.

Consuming Messages

Create a consumer by inheriting ConsumerServiceBase (for raw messages and errors):

using Library.Kafka;
using Confluent.Kafka;

public class EmailMessageConsumer : ConsumerServiceBase
{
    public override void OnMessageDelivered(string message)
    {
        var email = Newtonsoft.Json.JsonConvert.DeserializeObject<EmailMessage>(message);
        // handle email delivery logic here
    }

    public override void OnErrorOccured(Error error)
    {
        // handle/log error as needed
    }
}

Then register and activate each consumer in your DI container:

builder.Services.AddKafkaConsumer<EmailMessageConsumer>("topic1");

Consumers run as background hosted services.

Dependency Injection Patterns

• Producers: Scoped as Singleton, injectable anywhere (Controller, Service, etc).
• Consumers: Registered per topic, run as IHostedService background tasks.
• Extensible error & message handling: Consumers override OnMessageDelivered and OnErrorOccured.

Advanced Usage

Multiple Topics & Consumers

builder.Services.AddKafkaConsumer<Consumer1>("topic1");
builder.Services.AddKafkaConsumer<Consumer2>("topic2");
// as many as needed

Typed Message Consumption

You may also build type-safe consumers by using message base generics:

using Library.Kafka;
using Newtonsoft.Json;

public class StrongTypedConsumer : ConsumerServiceBase
{
    public override void OnMessageDelivered(string message)
    {
        var msgObj = JsonConvert.DeserializeObject<MyCustomMessage>(message);
        // Now process the strongly-typed object...
    }
    public override void OnErrorOccured(Error error)
    {
        // Handle error
    }
}

Error Handling

Override OnErrorOccured for logging, retry, or alerts:

public override void OnErrorOccured(Error error)
{
    Console.WriteLine($"Kafka Error: {error.Reason}");
}

API Reference

Producer

• void Produce(string topic, object message)
  • Serializes any object as JSON, sends to topic.

Consumer

• Inherit from ConsumerServiceBase
  • void OnMessageDelivered(string message)
  • void OnErrorOccured(Error error)

Registration

• services.AddKafkaProducer()
• services.AddKafkaConsumer<T>(string topic) — registers T as consumer for topic.

Authors

• Mehmet Yoldaş (mehmet-yoldas)

See all contributors on NuGet.

Acknowledgements

• Built on Confluent.Kafka and Newtonsoft.Json
• Thanks to the community and open source supporters

For issues, contributions or feedback, see mehmet-yoldas/library-core