IronAlpine.Framework.EventBus 3.0.5

IronAlpine.Framework.EventBus

Kafka event bus with a transactional outbox, an idempotent inbox, retries with backoff, dead letter tables, replay, retention cleanup and readiness health - all stored in the service's own EF Core database.

• Package id: IronAlpine.Framework.EventBus (3.0.5, net10.0)
• Depends on: Kernel, Mediator, Data, Confluent.Kafka 2.12

How it works

publisher service                                   consumer service
-----------------                                   ----------------
handler / domain event handler                      KafkaInboxWriterHostedService
  IOutboxStore.AddAsync(event)  --> OutboxMessages     consumes subscribed topics
                                      |                writes InboxMessages (idempotency key = event id)
KafkaOutboxPublisherHostedService     |                commits the Kafka offset
  claims pending rows (lease) --------+--> Kafka --->
  produces (acks=all, idempotent)                   InboxProcessorHostedService
  marks processed / retry / dead letter               claims pending rows (lease)
                                                      deserializes to the registered CLR type
                                                      IMediator.Publish(event) -> INotificationHandler<TEvent>
                                                      marks processed / retry / dead letter

Delivery is at least once end to end. Consumers must be idempotent.

Installation

<PackageReference Include="IronAlpine.Framework.EventBus" Version="3.0.5" />

Registration

using IronAlpine.EventBus.Extensions;

builder.Services.AddIronAlpineData<OrdersDbContext>(builder.Configuration, cfg => /* ... */);

builder.Services.AddIronAlpineEventBusKafka(
    builder.Configuration,
    options => options.ConsumerGroupId = "orders-service",            // optional code overrides
    topology =>
    {
        topology.Publish<OrderPlacedIntegrationEvent>("orders.lifecycle");
        topology.Publish<OrderCancelledIntegrationEvent>("orders.lifecycle");
        topology.Consume<PaymentCapturedIntegrationEvent>("payments.lifecycle");
    });

Overloads: AddIronAlpineEventBusKafka(configuration, configure, configureTopology, registerStorage = true, enableCleanupHostedService = true) and AddIronAlpineEventBusKafka(configuration, configureTopology, ...). (AddIronAlpineEventbusKafka with a lower-case b is obsolete.)

It registers:

The stores use the scoped DbContext alias registered by AddIronAlpineData. Add a migration after registering the event bus; design-time factories must call AddIronAlpineEventBusKafka too, or the tables are missing.

Start-up validation (when Enabled): consume routes require Writer.Enabled and Processor.Enabled; publish routes require Outbox.Enabled. The legacy section IronAlpine:EventBus:Kafka:Kafka throws.

Configuration

{
  "IronAlpine": {
    "EventBus": {
      "Kafka": {
        "Enabled": true,
        "BootstrapServers": "kafka:9092",
        "ConsumerGroupId": "orders-service",
        "ClientId": "orders-service",
        "SecurityProtocol": "SaslSsl",
        "SaslMechanism": "ScramSha512",
        "SaslUsername": "orders",
        "SaslPassword": "<secret>",
        "SslCaLocation": "/certs/ca.crt",
        "SslCertLocation": null,
        "SslKeyLocation": null,
        "SslKeyPassword": null,
        "SslEndpointIdentificationAlgorithm": "Https",
        "EnableAutoCommit": false,
        "EnableAutoOffsetStore": false,
        "AutoOffsetReset": "Earliest",
        "Provisioning": { "Enabled": true, "DefaultPartitions": 3, "DefaultReplicationFactor": 1, "MetadataWaitTimeoutSeconds": 20, "MetadataPollIntervalMilliseconds": 500 },
        "Writer": { "Enabled": true, "StartupDelaySeconds": 5, "PollTimeoutMilliseconds": 1000, "IdleDelayMilliseconds": 100, "WorkerErrorDelaySeconds": 2 },
        "Processor": { "Enabled": true, "BatchSize": 64, "LockSeconds": 30, "IdleDelayMilliseconds": 100, "WorkerErrorDelaySeconds": 2, "MaxRetryCount": 100, "FastRetryCount": 3, "FastRetryDelaySeconds": 5, "BackoffBaseDelaySeconds": 15, "MaxRetryDelaySeconds": 500 },
        "Outbox": { "Enabled": true, "BatchSize": 64, "LockSeconds": 30, "IdleDelayMilliseconds": 100, "WorkerErrorDelaySeconds": 2, "MaxRetryCount": 50, "FastRetryCount": 3, "FastRetryDelaySeconds": 5, "BackoffBaseDelaySeconds": 15, "MaxRetryDelaySeconds": 500, "MaxMessageBytes": 950000 },
        "Cleanup": { "Enabled": true, "ProcessedRetentionDays": 14, "DeadLetterRetentionDays": 60, "ReplayAuditRetentionDays": 180, "IntervalHours": 24, "BatchSize": 1000 },
        "Schema": {
          "InboxTableName": "InboxMessages",
          "OutboxTableName": "OutboxMessages",
          "InboxDeadLetterTableName": "InboxDeadLetters",
          "OutboxDeadLetterTableName": "OutboxDeadLetters",
          "ReplayOperationTableName": "EventReplayOperations",
          "RequireIdempotencyUniqueIndex": true,
          "RequireConcurrencyToken": true
        }
      }
    }
  }
}

Values above are the defaults except for the connection/security examples. Notes:

• ConsumerGroupId defaults to ia-eventbus-inbox (EventBusKafkaOptions.DefaultConsumerGroupId). Set a unique group per service: services sharing a group split each topic's messages between them. A service with consume routes logs a start-up warning while it uses the default (the default itself is unchanged for compatibility).
• SecurityProtocol / SaslMechanism / SslEndpointIdentificationAlgorithm are Confluent enum names (Plaintext, Ssl, SaslPlaintext, SaslSsl; Plain, ScramSha256, ScramSha512, ...; None, Https). SASL protocols require username and password; SSL protocols require SslCaLocation.
• Schema.*MaxLength options set column lengths (MessageTypeMaxLength 512, TopicMaxLength 256, KeyMaxLength 256, IdempotencyKeyMaxLength 256, LastErrorMaxLength 4000, LastErrorDetailMaxLength 8000, ...).
• The producer uses EnableIdempotence = true, Acks = All, MessageTimeoutMs = 30000. The consumer commits offsets manually after the inbox row is stored.

Integration events

using IronAlpine.EventBus.Contracts;
using IronAlpine.Kernel.Primitives;

public sealed record OrderPlacedIntegrationEvent(Guid OrderId, Guid CustomerId, decimal Total)
    : IntegrationEventBase, INotification;

IntegrationEventBase sets EventId, OccurredOnUtc, EventType (CLR full name) and EventVersion = 1, and exposes setters: SetEventId, SetOccurredOnUtc, SetEventType, SetCorrelationId, SetCausationId, SetPartitionKey, SetEventVersion.

Implement INotification on every event a service consumes: the inbox processor dispatches with IMediator.Publish, which rejects non-notifications (the message is retried and finally dead-lettered).

Wire format:

• Kafka key: PartitionKey, else the event id (N format). Set a partition key (for example the aggregate id) to keep the order of events of one entity.
• value: the event serialized with System.Text.Json web defaults; eventType, correlationId, causationId, partitionKey are rewritten in the payload;
• headers: correlation-id, causation-id, traceparent, tracestate, baggage, partition-key, event-id, event-type, event-version. Correlation id falls back to the current Activity trace id, then the event id.

Topology

topology.Publish<OrderPlacedIntegrationEvent>("orders.lifecycle");
topology.Publish<OrderPlacedIntegrationEvent>("orders.lifecycle", new EventTopicSpecification(Partitions: 6, ReplicationFactor: 3));
topology.Consume<PaymentCapturedIntegrationEvent>("payments.lifecycle", eventTypeName: "Payments.PaymentCaptured");
topology.Consume(typeof(PaymentRefundedIntegrationEvent), "payments.lifecycle");   // EventTopologyBuilderTypeExtensions

• The event type name defaults to the CLR full name (assembly-qualified suffixes are stripped). Publisher and consumer must agree on it: share the contract assembly, or pass the same eventTypeName on both sides.
• One CLR type maps to exactly one topic and one type name; one type name maps to one CLR type; topic specifications must be identical wherever a topic is declared. Violations throw when the registry is built.
• Many event types may share a topic; consumers route by (topic, event type).
• With Provisioning.Enabled the writer and publisher create missing topics and wait for metadata. An existing topic whose partition count or replication factor differs from the declared/default values is used as is and a warning is logged (up to 3.0.4 this threw out of the hosted service and stopped the host). Align the declared values with the real topic to silence the warning.

Publishing

public sealed class OrderPlacedDomainEventHandler(IOutboxStore outbox) : INotificationHandler<OrderPlacedDomainEvent>
{
    public async Task Handle(OrderPlacedDomainEvent notification, CancellationToken cancellationToken)
    {
        var integrationEvent = new OrderPlacedIntegrationEvent(notification.OrderId, notification.CustomerId, notification.Total);
        integrationEvent.SetPartitionKey(notification.OrderId.ToString());
        await outbox.AddAsync(integrationEvent, cancellationToken);   // no SaveChanges here
    }
}

• IOutboxStore.AddAsync(IIntegrationEvent) builds the envelope (the type must be registered with Publish) and adds an OutboxMessageEntity.
• Inside domain event dispatch (DomainEventDispatchScope.IsActive) it does not save: the row is written by the SaveChangesAsync that dispatched the domain event, atomically with the aggregate change.
• Outside that scope it calls SaveChangesAsync immediately, which also saves every other pending change of the DbContext. Call it after the business changes are complete, or publish from a domain event handler.
• IEventBus.PublishAsync(event) does the same through IOutboxStore.

Consuming

public sealed class PaymentCapturedHandler(IRepository<Order> orders, IUnitOfWork unitOfWork)
    : INotificationHandler<PaymentCapturedIntegrationEvent>
{
    public async Task Handle(PaymentCapturedIntegrationEvent message, CancellationToken cancellationToken)
    {
        var order = await orders.GetByIdAsync(message.OrderId, cancellationToken);
        if (order is null || order.IsPaid)
        {
            return;                       // idempotent: duplicates and replays are no-ops
        }

        order.MarkPaid(message.OccurredOnUtc);
        await unitOfWork.SaveChangesResilientAsync(cancellationToken);
    }
}

• Each message is published in a new DI scope; all handlers of the event run sequentially in that scope.
• Mediator pipeline behaviours (logging, validation, transaction) do not run for these handlers; each SaveChanges is its own transaction.
• A handler exception schedules a retry: attempts 1..FastRetryCount wait FastRetryDelaySeconds, later attempts wait BackoffBaseDelaySeconds * 2^n capped at MaxRetryDelaySeconds. After MaxRetryCount the message is dead-lettered with RETRY_EXHAUSTED. The exception type does not matter (any exception is retried).
• Other dead letter reasons (ReasonCodes): TYPE_NOT_REGISTERED (no Consume route for topic + type), TYPE_MISMATCH (header and payload types differ), DESERIALIZATION_ERROR, TRANSPORT_NON_RETRIABLE (outbox: message too large or non-retriable produce error).
• If several handlers exist and one fails, the others may already have saved; on retry all run again.
• Retries and parallel workers do not preserve ordering. Use OccurredOnUtc / versions in handlers to ignore stale events.

Dead letters and replay

var page = await operationReader.GetDeadLettersAsync(new DeadLetterQuery
{
    Direction = DeadLetterDirection.Inbox,
    EventType = typeof(PaymentCapturedIntegrationEvent).FullName,
    PageNumber = 1,
    PageSize = 50                                   // max 200
}, ct);

var result = await replayService.ReplayInboxDeadLettersAsync(new ReplayRequest
{
    EventType = typeof(PaymentCapturedIntegrationEvent).FullName,
    OccurredFromUtc = DateTime.UtcNow.AddDays(-1),  // filters on FailedOnUtc
    MaxItems = 100,                                 // default 1000, max 10000
    DryRun = false,
    RequestedBy = userName,
    Reason = "handler fixed",
    Source = "admin-api"
}, ct);

Replay re-queues matching dead letters and records an EventReplayOperations audit row (also for dry runs). The dead-lettered inbox/outbox row is kept until ProcessedRetentionDays and still owns the idempotency key, so replay re-arms that row (pending again, retry count and errors cleared) instead of inserting a duplicate key; only when the row no longer exists is a new pending row inserted. Several dead letters of the same event re-queue it once. The dead letter rows are kept. IMessageOperationReader also lists replay operations (GetReplayOperationsAsync(EventReplayOperationQuery)). The service has its own PagedResult<T> type (IronAlpine.EventBus.Contracts), distinct from the Kernel one.

Health and diagnostics

• IEventBusHealthState: TopologyReady, TransportReady, writer/processor heartbeats, LastError, package version and assembly hash. ia-eventbus-ready reports Unhealthy (topology/transport not ready), Degraded (recent error) or Healthy; it is always Healthy while Enabled = false. It is tagged ready, so /health/ready from MapIronAlpineHealthChecks (Web package) includes it (since 3.0.5). It can still be mapped on its own:

  app.MapHealthChecks("/health/eventbus", new HealthCheckOptions { Predicate = r => r.Name == "ia-eventbus-ready" });
  

• Activities are emitted from the ActivitySource named IronAlpine.Eventbus (eventbus.outbox.publish, eventbus.inbox.write, eventbus.inbox.process). Add the source to your tracer provider to export them.

• Logs use the [IA EventBus] prefix.

Cleanup

EventBusCleanupHostedService runs every Cleanup.IntervalHours and deletes in batches: processed inbox/outbox rows older than ProcessedRetentionDays, dead letters older than DeadLetterRetentionDays, replay audit rows older than ReplayAuditRetentionDays (minimum 1 day each). Disable it with enableCleanupHostedService: false or Cleanup.Enabled = false.

Extension points

• Replace any store (IOutboxStore, IInboxStore, IDeadLetterStore, IReplayService, IMessageOperationReader) by registering it before AddIronAlpineEventBusKafka (registrations use TryAdd), or pass registerStorage: false and register storage yourself.
• EventEnvelopeFactory.Create(payload, registry) and EventTypeCanonicalizer are public for custom transports.
• ModelBuilderExtensions.ApplyIronAlpineEventbus(modelBuilder, schema) maps the tables manually.

Pitfalls

• Consumed event types must implement INotification.
• Renaming or moving an event class changes its default type name; consumers then dead-letter it as TYPE_NOT_REGISTERED. Pin eventTypeName before refactoring contracts.
• IOutboxStore.AddAsync outside domain event dispatch saves the whole DbContext.
• In a publish-only service (no Consume routes) the inbox writer stays idle and does not touch TransportReady (up to 3.0.4 it kept marking the transport unavailable between publishes).
• ia-eventbus-ready is part of /health/ready since 3.0.5: a service whose transport or topology is not ready is reported not ready.
• The outbox claim order is by CreatedOnUtc, the inbox claim order by ReceivedOnUtc; there is no per-key ordering guarantee once retries happen.