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