Processing directly from Kafka topics

Nexus allows you to use Kafka as an external pending storage, which means that Nexus reads and processes messages directly from Kafka without first storing them in an SQL backed Nexus queue.

Use it by first installing CommerceMind.Nexus.Kafka and then configuring the external storage:

builder.Services
    .AddNexus()
    .AddKafkaExternalPendingStorage(options =>
    {
        options.BootstrapServers = "localhost:9092";
        options.ConsumerGroupId = "my-app-processing";
        options.ReadinessConsumerGroupId = "my-app-readiness";
    });

And then on your queue:

[Queue("export_order", StorageMode = QueueStorageMode.ExternalPending)]
public class ExportOrderQueueMessage : IQueueMessageWithId
{ }

The processing consumer and readiness bridge read the same topic through different consumer groups. The processing group owns leasing and completion. The readiness group observes every record independently and commits its offset only after activation is durable. The two group IDs must be different.

With the default prefix, queue export_order maps to topic nexus-export_order. Direct producers publish the Nexus external-pending envelope to that topic; no companion message is needed. Existing topics can be used after configuring the prefix and distinct groups, and the readiness group will catch up from its own committed offsets.

Read more about external pending storage here.

Enqueueing from Kafka topics

Nexus has built-in support to use the very popular Apache Kafka platform for both instance events as well as taking messages from a Kafka topic and enqueueing into a Nexus queue.

In order to use this you need to install the NuGet package called CommerceMind.Nexus.Kafka.

You set it up like this:

builder.Services
    .AddNexus()
    .AddQueues()
    .AddKafka(options =>
    {
        options.ConsumerConfig.BootstrapServers = "...";
        options.ConsumerConfig.GroupId = "...";
    })
    .AddNexusKafkaQueues(options => 
    {
        // This allows you to control things like which status or when messages are processed
        options.EnqueueContextFactory = message => new EnqueueContext { Priority = message.Id != null ? 10 : 0 }
    })
        .EnqueueFrom<KafkaQueueMessage>("kafka-topic");

With this examples all messages posted to the kafka-topic will get placed in the Nexus queue for KafkaQueueMessage.

Instance events

You can also use Kafka as the way to send signals between Nexus instances in a scaled out environment.

builder.Services
    .AddNexus()
    .AddQueues()
    .AddKafka(options =>
    {
        options.ConsumerConfig.BootstrapServers = "...";
        options.ConsumerConfig.GroupId = "...";

        options.ProducerConfig.BootstrapServers = "...";
    })
    .AddKafkaInstanceEvents(options => 
    {
        // Change this to something unique if you're using the same Kafka instance for multiple
        // different Nexus applications
        options.TopicName = "nexus_instance_events";
    });