DotKafka 2.0.1

DotKafka - A .NET Kafka SDK

A lightweight .NET SDK for Apache Kafka, inspired by Spring Kafka, providing annotation-based consumers and a simple producer interface.

Features

• Attribute-based Kafka listeners ([KafkaListener]) for consumers.
• IKafkaProducer interface for sending messages.
• KafkaTemplate<TKey, TValue> for flexible message production.
• Automatic topic creation support.
• Integration with Microsoft.Extensions.DependencyInjection.
• Support for retry mechanisms and Dead Letter Queues (DLQ).

Installation

Install via NuGet:

dotnet add package dotkafka

Usage

Consumer Example

public class EventConsumer
{
    [KafkaListener(topics: "sme.analytics", groupId: "default-group")]
    public async Task HandleAsync(ElectoralData message)
    {
        Console.WriteLine($"Consumed message: {message}");
        await Task.CompletedTask;
    }

    [KafkaListener(topics: "analytics", groupId: "default-group", isBatch: true)]
    public async Task HandleAnalyticsAsync(List<ElectoralData> message)
    {
        Console.WriteLine($"Consumed message: {message}");
        await Task.CompletedTask;
    }

    [KafkaListener(topics: new[] { "user.actions" }, groupId: "user-group", isBatch: true)]
    [Retryable(maxRetries: 3, retryDelayMs: 2000, deadLetterTopic: "user.events.dlq")]
    public async Task HandleBatchUserEventsAsync(List<ElectoralData> messages)
    {
        await Task.CompletedTask;
        throw new Exception("Simulated processing failure");
    }

    [KafkaListener(pattern: "analytics.*", groupId: "analytics-group")]
    public async Task HandleAnalyticsEventAsync(ElectoralData message)
    {
        await Task.CompletedTask;
    }

    [KafkaListener(pattern: "telco.*", groupId: "logs-group", isBatch: true)]
    [Retryable(maxRetries: 2, retryDelayMs: 1000)]
    public async Task HandleLogsBatchAsync(List<ElectoralData> messages)
    {
        await Task.CompletedTask;
    }
}

Producer Example

public class EventProducer
{
    private readonly IKafkaProducer _kafkaProducer;

    public EventProducer(IKafkaProducer kafkaProducer)
    {
        _kafkaProducer = kafkaProducer;
    }

    public async Task SendMessageAsync(string topic, ElectoralData message)
    {
        await _kafkaProducer.ProduceAsync(topic, message);
    }
}

Using KafkaTemplate

KafkaTemplate<TKey, TValue> provides a simple way to send messages with optional headers, partitions, and timestamps.

public class EventService
{
    private readonly KafkaTemplate<string, ElectoralData> _kafkaTemplate;

    public EventService(KafkaTemplate<string, ElectoralData> kafkaTemplate)
    {
        _kafkaTemplate = kafkaTemplate;
        _kafkaTemplate.DefaultTopic = "default-topic";
    }

    public async Task SendEventAsync(ElectoralData data)
    {
        await _kafkaTemplate.SendDefaultAsync(data);
    }

    public async Task SendWithKeyAsync(string key, ElectoralData data)
    {
        await _kafkaTemplate.SendDefaultAsync(key, data);
    }

    public async Task SendToPartitionAsync(int partition, string key, ElectoralData data)
    {
        await _kafkaTemplate.SendDefaultAsync(partition, key, data);
    }

    public async Task SendWithTimestampAsync(int partition, DateTime timestamp, string key, ElectoralData data)
    {
        await _kafkaTemplate.SendDefaultAsync(partition, timestamp, key, data);
    }

    public async Task SendToTopicAsync(string topic, string key, ElectoralData data)
    {
        await _kafkaTemplate.SendAsync(topic, key, data);
    }

    public async Task SendSimpleAsync(string topic, ElectoralData data)
    {
        await _kafkaTemplate.SendAsync(topic, data);
    }

    public async Task SendToSpecificPartitionAsync(string topic, int partition, string key, ElectoralData data)
    {
        await _kafkaTemplate.SendAsync(topic, partition, key, data);
    }
}

Web API Controller Example

[ApiController]
[Route("[controller]")]
public class AnalyticsController : ControllerBase
{
    private readonly ILogger<AnalyticsController> _logger;
    private readonly IKafkaTemplate<string, string> _template;

    public AnalyticsController(ILogger<AnalyticsController> logger, IKafkaTemplate<string, string> template)
    {
        _logger = logger;
        _template = template;
    }

    [HttpPost("send", Name = nameof(SendAnalyticsData))]
    public async Task<ActionResult<object>> SendAnalyticsData([FromBody] ElectoralData request)
    {
        var data = JsonConvert.SerializeObject(request);
        await _template.SendAsync("user.events", data);
        return Ok();
    }
}

Dependency Injection Setup

services.AddScoped<EventConsumer>();
services.AddScoped<KafkaTemplate<string, ElectoralData>>();
services.AddKafkaListener(config =>
{
    config.BootstrapServers = "localhost:9092";
    config.AutoOffsetReset = Confluent.Kafka.AutoOffsetReset.Earliest;
    config.EnableAutoCommit = true;
    config.TimeoutInMilliseconds = 5000;
});

builder.Services.AddKafkaTemplate<string, string>(c =>
{
    c.BootstrapServers = "localhost:9092";
});

Producer Registration

services.AddDotKafkaProducer(config =>
{
    config.BootstrapServers = "localhost:9092";
});

License

This project is licensed under the MIT License.

Contributing

Contributions are welcome! Feel free to open an issue or submit a pull request.

Contact

For support or inquiries, please reach out to the project maintainers.