Luciano Azevedo

Posted on Oct 18

Consuming Kafka events as background jobs in .NET with NexJob

#dotnet #csharp #kafka #backgroundjobs

When a message arrives on a Kafka topic, processing it directly inside your consumer loop has a well-known problem: if the handler is slow or makes external network calls, you risk exceeding max.poll.interval.ms and causing a partition rebalance. If the handler fails halfway through, you either commit the offset and lose the work, or crash the consumer loop.

A standard pattern to solve this is to decouple ingestion from execution: read the message, persist it immediately as a background job, commit the Kafka offset, and let worker slots process it with retries and dead-lettering.

This post sets up a Kafka consumer in .NET that turns incoming topic messages into NexJob background jobs.

What you need

The .NET 8 or 10 SDK
A running Kafka broker (e.g., localhost:9092)

1. A project and packages

Create a worker service or API project:

dotnet new worker -n KafkaWorker
cd KafkaWorker
dotnet add package NexJob
dotnet add package NexJob.Kafka

NexJob.Kafka bundles both the consumer trigger and the outbox producer in a single library.

2. Write the job

A consumer trigger hands the raw message payload to a job implementing IJob<string>.

using System.Text.Json;
using NexJob;

public sealed record OrderPlacedEvent(Guid OrderId, decimal Amount, string CustomerEmail);

public sealed class ProcessOrderJob : IJob<string>
{
    private readonly ILogger<ProcessOrderJob> _logger;

    public ProcessOrderJob(ILogger<ProcessOrderJob> logger) => _logger = logger;

    public Task ExecuteAsync(string payload, CancellationToken ct)
    {
        var order = JsonSerializer.Deserialize<OrderPlacedEvent>(payload);
        if (order is null)
        {
            throw new InvalidOperationException("Failed to deserialize order payload.");
        }

        _logger.LogInformation("Processing order {OrderId} for {Amount:C}", order.OrderId, order.Amount);

        // Your business logic here: charge card, update database, call API
        return Task.CompletedTask;
    }
}

The job is resolved through dependency injection, so you can inject your DbContext, HTTP clients, or repositories into the constructor.

3. Register the Kafka trigger

In Program.cs, bind the topic to your job:

using NexJob;
using NexJob.Kafka;

var builder = Host.CreateApplicationBuilder(args);

// Register NexJob engine and the Kafka trigger
builder.Services.AddNexJob(opt =>
    {
        opt.Workers = 5; // Concurrent worker threads processing jobs
    })
    .AddKafkaTrigger<ProcessOrderJob>(opt =>
    {
        opt.BootstrapServers = "localhost:9092";
        opt.Topic = "orders";
        opt.GroupId = "orders-processor-group";
    })
    .AddNexJobJobs(typeof(Program).Assembly);

var host = builder.Build();
await host.RunAsync();

AddKafkaTrigger<TJob> registers the consumer as a background service and automatically registers ProcessOrderJob as transient in your service container.

4. How offsets and retries work

A trigger does not execute the job directly inside the Kafka loop. The lifecycle works as follows:

Consume: The consumer reads a message batch from Kafka.
Store: Each message is enqueued into NexJob storage. The Kafka message key (or partition + offset) is used as the idempotency key, preventing duplicate jobs if a network glitch occurs.
Commit: The offset is committed to Kafka only after the job is safely written to storage.
Execute: A NexJob worker picks up the job from storage and runs ExecuteAsync.

If your business logic throws an exception inside ExecuteAsync:

The Kafka partition is not blocked; other messages continue to be consumed.
The job follows your NexJob retry policy (exponential backoff by default).
If retries are exhausted, the job moves to the dead-letter queue.

5. Publishing messages back to Kafka (Outbox pattern)

If your app also needs to produce messages reliably, NexJob.Kafka includes an outbox producer. Instead of making an in-flight network call to Kafka inside an HTTP request or database transaction, enqueue it through IScheduler:

app.MapPost("/orders", async (OrderPlacedEvent order, IScheduler scheduler) =>
{
    // Save to local database first, then enqueue Kafka message
    await scheduler.EnqueueKafkaAsync(
        topic: "orders",
        key: order.OrderId.ToString(),
        value: order);

    return Results.Accepted();
});

NexJob persists the message locally and delivers it to Kafka in the background with retry and guaranteed ordering per key.

Next steps

Using a database provider (NexJob.Postgres or NexJob.SqlServer) instead of in-memory storage for persistent guarantees across restarts.
Setting dead-letter alert topics with options.ExhaustedJobsTopic.
Monitoring consumers and worker slots using OpenTelemetry.

Source and samples: https://github.com/oluciano/NexJob
Documentation: https://oluciano.github.io/NexJob/

DEV Community