NexJob.Kafka 6.7.0

NexJob.Kafka

End-to-end Apache Kafka integration for NexJob, unifying Kafka Triggers (Consumers) and a Resilient Outbox Producer in a single package.

Installation

dotnet add package NexJob.Kafka

1. Resilient Outbox Producer

Publishes messages to Kafka topics backed by NexJob's persistent storage, retry with backoff and jitter, dead-letter dispatch, and OpenTelemetry trace propagation.

Registration

using NexJob;
using NexJob.Kafka;

builder.Services.AddNexJobPostgres(builder.Configuration.GetConnectionString("NexJobConnection")!); // or any other storage provider
builder.Services.AddNexJob()
    .AddKafkaProducer(options =>
    {
        options.BootstrapServers = builder.Configuration["KAFKA_BOOTSTRAP_SERVERS"] 
            ?? Environment.GetEnvironmentVariable("KAFKA_BOOTSTRAP_SERVERS") 
            ?? "localhost:9092";
        options.Acks = Confluent.Kafka.Acks.All;
        options.EnableIdempotence = true;
        options.FlushTimeout = TimeSpan.FromSeconds(10);
    });

Publishing Messages

Inject IScheduler into your services or controllers:

public class OrderService(IScheduler scheduler)
{
    // 1. Strongly typed object (automatically serialized to JSON)
    public async Task CreateOrderAsync(OrderCreatedEvent order, CancellationToken ct)
    {
        await scheduler.EnqueueKafkaAsync(
            topic: "orders-topic",
            key: order.OrderId.ToString(),
            value: order,
            cancellationToken: ct);
    }

    // 2. Raw string / JSON payload
    public async Task PublishRawJsonAsync(string orderId, string rawJson, CancellationToken ct)
    {
        await scheduler.EnqueueKafkaRawAsync(
            topic: "orders-topic",
            key: orderId,
            value: rawJson,
            cancellationToken: ct);
    }

    // 3. Raw binary payload
    public async Task PublishBinaryAsync(string key, byte[] bytes, CancellationToken ct)
    {
        await scheduler.EnqueueKafkaRawAsync(
            topic: "events-topic",
            key: key,
            value: bytes,
            cancellationToken: ct);
    }
}

2. Kafka Trigger (Consumer)

Consumes incoming messages from Kafka topics and automatically enqueues them as background jobs.

Registration

There are two ways to register which job is triggered when a message arrives:

Option A: Strongly-Typed Consumer (Recommended)

Bind a topic directly to a job handler class (IJob<string>). The job is automatically registered in DI as Transient:

builder.Services.AddNexJob()
    .AddKafkaTrigger<ProcessOrderJob>(options =>
    {
        options.BootstrapServers = builder.Configuration["KAFKA_BOOTSTRAP_SERVERS"] ?? "localhost:9092";
        options.Topic = "incoming-orders";
        options.GroupId = "nexjob-consumer-group";
        options.TargetQueue = "orders";
    });

Option B: Dynamic Message Header (nexjob.job_type)

If multiple job types share the same topic, omit the generic argument. The incoming Kafka message must include the nexjob.job_type header, unless options.JobType is set — when it is, it always wins over the header, so a producer cannot override which job runs:

builder.Services.AddNexJob()
    .AddKafkaTrigger(options =>
    {
        options.BootstrapServers = builder.Configuration["KAFKA_BOOTSTRAP_SERVERS"] ?? "localhost:9092";
        options.Topic = "incoming-orders";
        options.GroupId = "nexjob-consumer-group";
        options.TargetQueue = "orders";
        // options.JobType = typeof(DefaultOrderJob).AssemblyQualifiedName;
    });

Inbound Message Contract

For dynamic triggers (Option B), consumed messages can carry the following headers:

• nexjob.job_type: Assembly-qualified name of the IJob<string> to execute. Read only when options.JobType is unset — a configured JobType always wins over this header. A message with neither can never become a job and is handled as a permanent failure.
• traceparent: W3C distributed trace header (optional).

The message value is passed as the string input to the resolved job. The job idempotency key is the record position (kafka:{topic}:{partition}:{offset}), so a redelivered record never creates a second job while two records that share a message key each produce a job.

Enqueue failures are classified: a transient failure (storage or network error) retries the same record in place with a 1 s to 30 s backoff and commits only after success; a permanent failure goes to DeadLetterTopic and is committed (without a topic it is logged at Error and committed, so it cannot block the partition).

3. Configuration & 12-Factor App (Docker / Kubernetes)

NexJob.Kafka is configured through the option delegates shown above; it does not read appsettings.json sections or environment variables by itself. In containerized environments, read the variables in your own code and assign them to the options:

# Set environment variables in Docker / Kubernetes
export KAFKA_BOOTSTRAP_SERVERS="kafka-broker.prod:9092"
export KAFKA_TOPIC="orders"

Read them in C#:

builder.Services.AddNexJob()
    .AddKafkaProducer(options =>
    {
        options.BootstrapServers = Environment.GetEnvironmentVariable("KAFKA_BOOTSTRAP_SERVERS")
            ?? throw new InvalidOperationException("KAFKA_BOOTSTRAP_SERVERS is not set.");
    });

4. Configuration Options

Producer Options (KafkaProducerOptions)

Trigger Options (KafkaTriggerOptions)

5. Architectural Guarantees

1. Transactional Outbox / Resilient Publish: Messages are first durably persisted to NexJob storage. If the Kafka broker is down, messages wait safely in storage and are retried automatically.
2. Dead-Letter Handling: Permanent publishing failures trigger NexJob's dead-letter pipeline (IDeadLetterHandler) and surface in the dashboard.
3. Trace Propagation: Injects W3C traceparent headers into outgoing messages and extracts them on consumer triggers.
4. Graceful Shutdown: Unflushed in-flight messages are flushed before the application process exits.

6. Sagas & Event-Driven Choreographies

Need distributed state machines or sagas with compensating transactions over Kafka? See qKafka — the companion event-driven framework that pairs with NexJob.

Forwarding failed jobs

A job created from a consumed message that exhausts its retries stays only inside NexJob (Failed in the dashboard), because the broker message was acknowledged when it became a job. To copy the original message body to another topic, set ExhaustedJobsTopic on the trigger (the Outbox producer, AddKafkaProducer, is required):

.AddKafkaTrigger<ProcessOrderJob>(opt =>
{
    opt.Topic = "orders";
    opt.TargetQueue = "orders";
    opt.ExhaustedJobsTopic = "orders.exhausted";   // optional: opt.ExhaustedJobsIncludeErrorHeader = true
});

Only the body is forwarded, the job stays Failed in NexJob, and only jobs of this trigger are forwarded. Details and limits: Dead-letter forwarding.