EasyCore.Kafka 8.3.1

📨 EasyCore.Kafka

EasyCore.Kafka is a Kafka infrastructure client for .NET 8. Built on Confluent.Kafka, it provides DI registration, connect, topic publish/subscribe, and manual Commit. It is not tied to IEvent / EventBus — use it standalone, or via the EasyCore.EventBus.Kafka adapter.

🌍 Language

• Chinese: README.md
• English (this document)

📚 Table of Contents

• 1. Positioning
• 2. Relation to EventBus
• 3. Requirements
• 4. Installation
• 5. Quick Start
• 6. API Overview
• 7. Options (KafkaOptions)
• 8. FAQ
• 9. License

1. 🎯 Positioning

1.1 Design Principles

2. Relation to EventBus

┌─────────────────────────────┐
│  App code (any payload)     │
└──────────────┬──────────────┘
               │ IKafkaClient
               ▼
┌─────────────────────────────┐
│  EasyCore.Kafka (this)      │  ← infrastructure client
└──────────────┬──────────────┘
               │ Confluent.Kafka
               ▼
          Kafka Cluster

┌─────────────────────────────┐
│ EasyCore.EventBus.Kafka     │  ← adapter: IEvent / Handler
│   └── uses this client      │
└─────────────────────────────┘

3. ⚙ Requirements

4. 📥 Installation

dotnet add package EasyCore.Kafka

5. ⚡ Quick Start

5️⃣.1️⃣ Register DI

using EasyCore.Kafka;

builder.Services.AddEasyCoreKafka(o =>
{
    o.BootstrapServers = "localhost:9092";
    o.TopicName = "EasyCore.Topic";
    o.GroupId = "EasyCore.GroupId";
    o.MessageTimeoutMs = 10000;
    o.RequestTimeoutMs = 10000;
});

5️⃣.2️⃣ Publish

public sealed class TelemetryPublisher
{
    private readonly IKafkaClient _client;

    public TelemetryPublisher(IKafkaClient client) => _client = client;

    public async Task PublishAsync(string deviceId, CancellationToken ct = default)
    {
        await _client.ConnectAsync(ct);

        var body = Encoding.UTF8.GetBytes($$"""{"deviceId":"{{deviceId}}"}""");
        var headers = new Dictionary<string, byte[]>
        {
            ["x-source"] = Encoding.UTF8.GetBytes("iot-gateway")
        };

        await _client.PublishAsync(
            topic: "telemetry.raw",
            body: body,
            key: deviceId,
            headers: headers,
            cancellationToken: ct);
    }
}

5️⃣.3️⃣ Subscribe and Commit

public sealed class TelemetryConsumer : BackgroundService
{
    private readonly IKafkaClient _client;

    public TelemetryConsumer(IKafkaClient client) => _client = client;

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        await _client.ConnectAsync(stoppingToken);

        await _client.SubscribeAsync(
            topics: new[] { "telemetry.raw" },
            handler: async (msg, ct) =>
            {
                var json = Encoding.UTF8.GetString(msg.Body.Span);
                // handle business…

                if (msg.NativeResult is not null)
                    _client.Commit(msg.NativeResult);

                await Task.CompletedTask;
            },
            cancellationToken: stoppingToken);
    }
}

KafkaDeliveredMessage fields: Topic, Key, Body, Headers, NativeResult (pass to Commit).

6. 🧩 API Overview

IKafkaClient implements IAsyncDisposable — dispose on shutdown.

7. Options (KafkaOptions)

8. ❓ FAQ

Q: This package vs EasyCore.EventBus.Kafka?
A: Use the EventBus adapter for IEvent, handlers, and retries. Use this package for raw topic I/O.

Q: What do I pass to Commit?
A: Pass KafkaDeliveredMessage.NativeResult from the handler (the underlying consume result).

Q: What do TopicName / GroupId do when used standalone?
A: Naming conventions / suffixes. PublishAsync / SubscribeAsync still use the topics you pass in.

Q: Multiple bootstrap servers?
A: Yes — set BootstrapServers to e.g. kafka1:9092,kafka2:9092.

9. 📄 License

MIT — see the repository LICENSE, or the NuGet package metadata.

🤝 Contributing

Issues and PRs are welcome. After changing this package, verify EasyCore.EventBus.Kafka and related demos.