Teamforapps.Messaging.Kafka 1.0.10

Teamforapps.Messaging.Kafka

Kafka producer/consumer library for Teamforapps microservices. Wraps Confluent.Kafka with idempotent publishing, Polly retry, and an optional EF Core transactional outbox — wired up through a single AddKafkaMsgService(...) call.

NuGet package ID: Teamforapps.Messaging.Kafka | Target framework: .NET 10

Features

• IIntegrationEventPublisher — produce events with an idempotent singleton Confluent.Kafka producer
• IIntegrationEventHandler<T> — type-safe consumer handlers with per-message DI scope and Polly retry
• [Topic("name")] attribute — explicit topic mapping; fallback is lowercased class name with "Event" suffix stripped
• EF Core transactional outbox — write event and business data in the same DB transaction; background service drains to Kafka
• Single fluent DI registration: AddKafkaMsgService(...)

Quick start

1. Install

dotnet add package Teamforapps.Messaging.Kafka

2. Define an integration event

using Teamforapps.Messaging.Kafka.Abstractions;
using Teamforapps.Messaging.Kafka.Topics;

[Topic("profile.created")]
public sealed record ProfileCreated(Guid EventId, DateTime OccurredAt, Guid ProfileId, string UserId)
    : IIntegrationEvent;

IIntegrationEvent requires EventId: Guid and OccurredAt: DateTime.

3. Configure appsettings.json

"Kafka": {
  "BootstrapServers": "localhost:29092",
  "ClientId": "profile-api",
  "ConsumerGroup": "profile-api"
}

All values can also be set via environment variables (e.g. Kafka__BootstrapServers).

4. Register in Program.cs

Producer only:

builder.Services.AddKafkaMsgService(builder.Configuration);

Consumer:

builder.Services.AddKafkaMsgService(builder.Configuration, b => b
    .Subscribe<ProfileCreated, IndexProfileOnCreated>()
    .Subscribe<ProfileDeleted, RemoveProfileFromIndex>());

Producer + outbox:

builder.Services.AddKafkaMsgService(builder.Configuration, b =>
    b.AddOutbox<ApplicationDbContext>());

Inline options (no config file):

builder.Services.AddKafkaMsgService(o =>
{
    o.BootstrapServers = "localhost:29092";
    o.ClientId = "profile-api";
    o.ConsumerGroup = "profile-api";
});

5. Publish an event

public class CreateProfileHandler(IIntegrationEventPublisher publisher)
{
    public async Task HandleAsync(CreateProfileCommand cmd, CancellationToken ct)
    {
        // ... business logic ...
        await publisher.PublishAsync(new ProfileCreated(Guid.NewGuid(), DateTime.UtcNow, profile.Id, cmd.UserId), ct);
    }
}

To publish to a specific partition, pass an explicit key:

await publisher.PublishAsync(evt, partitionKey: profile.Id.ToString(), ct);

6. Handle an event

public sealed class IndexProfileOnCreated : IIntegrationEventHandler<ProfileCreated>
{
    public async Task HandleAsync(ProfileCreated e, CancellationToken ct)
    {
        // index the profile in search
    }
}

Handlers are registered as scoped services, so they can depend on DbContext, unit-of-work, or any other scoped dependency.

Outbox pattern

When AddOutbox<TContext>() is called, IIntegrationEventPublisher is replaced with a version that writes an outbox_messages row inside the current EF transaction. A background service (OutboxPublisherHostedService) polls the table and produces to Kafka.

Add to OnModelCreating:

protected override void OnModelCreating(ModelBuilder modelBuilder)
{
    modelBuilder.ConfigureOutbox();
}

Generate the migration:

dotnet ef migrations add AddOutbox -p src/YourService.Infrastructure -s src/YourService.API

Configuration reference

All sections are optional — defaults are shown.

Kafka (root)

Kafka:Producer

Kafka:Consumer

Kafka:Outbox

Consumer retry behaviour

Failed handlers are retried with exponential backoff (RetryCount attempts, base delay RetryBaseDelay). After all retries are exhausted the consumer stops (fail-fast). A dead-letter queue strategy is planned for v2.

Key abstractions

Local development

# Start Kafka + AKHQ UI
cd deployments && docker compose up -d kafka akhq
# AKHQ UI: http://localhost:8080
# Kafka:   localhost:29092

# Run tests
dotnet test tests/Myn.Messaging.Kafka.Tests/

# Pack NuGet
dotnet pack src/Teamforapps.Messaging.Kafka -c Release -o ./nupkgs

License

MIT