IronAlpine.Framework.EventBus 3.0.6

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.6, 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.6" />

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, "OutboxProcessedRetentionDays": 0, "InboxProcessedRetentionDays": 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, CorrelationIdMaxLength 128, CausationIdMaxLength 128, ...).
• 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, SetTargetConsumer.

Event time = save instant

OccurredOnUtc defaults to SaveInstantScope.Current ?? DateTime.UtcNow (since 3.0.6). While SaveChangesAsync dispatches domain events, SaveInstantScope.Current is the save instant of that save - the exact value the Data package writes to CreatedOnUtc/ModifiedOnUtc (see the Data README, "Save instant"). So an event created in a domain event handler carries the same DateTime as the aggregate's audit stamp, and so does its outbox row.

This lets a consumer check whether a projection is in sync with the source: the source aggregate's ModifiedOnUtc ?? CreatedOnUtc equals the OccurredOnUtc of its last full-state event, which the projection stores (IOccurredOnUtc). Rule: create full-state events in a domain event handler (inside SaveChanges). An event created outside a save takes the current time; SetOccurredOnUtc(...) always wins over both.

Targeted delivery (TargetConsumer)

var repair = new OrderSnapshotIntegrationEvent(order.Id, order.Status, order.Total);
repair.SetTargetConsumer("billing-service");   // the target service's Kafka ConsumerGroupId
await outbox.AddAsync(repair, ct);

• The value travels as the target-consumer header (EventEnvelopeFactory.TargetConsumerHeader) and in the payload.
• The inbox writer of every other consumer group commits the offset and does not write the message to its inbox (Debug log). The comparison with ConsumerGroupId is ordinal (Kafka group ids are case-sensitive).
• Without the header (the default) every subscriber processes the event as before.
• The handler sees TargetConsumer on the deserialized event, so it can tell a targeted (for example repair) event from a regular one.
• IIntegrationEvent.TargetConsumer has a default implementation returning null, so custom implementations of the interface keep compiling.

Deserialization on the consumer

The base properties EventId, OccurredOnUtc, CorrelationId, CausationId, PartitionKey, EventVersion and TargetConsumer are marked [JsonInclude], so the event a handler receives carries the publisher's values. Up to 3.0.5 their private setters were ignored by System.Text.Json: a consumer saw a new EventId, its own receive time as OccurredOnUtc and empty correlation fields. EventType keeps the consumer's CLR name.

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, and target-consumer when set. 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 (OccurredOnUtc, Key, CorrelationId, CausationId and the full HeadersJson).
• 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).
• Unregistered event types are skipped (since 3.0.6): a message on a subscribed topic whose type has no Consume route for that topic is marked processed without calling a handler and without a dead letter (Warning log with EventType, Topic, MessageId). Many event types may share a topic (for example one topic per aggregate) and each consumer registers only the types it needs. Up to 3.0.5 such messages were dead-lettered as TYPE_NOT_REGISTERED; that reason code is removed. While the topology is not yet ready the message is retried as before. Only messages whose type name is known are skipped: a message without any event type name in its headers or payload is malformed and is dead-lettered as TYPE_UNRESOLVED.
• Other dead letter reasons (ReasonCodes): TYPE_UNRESOLVED (no event type name in headers or payload), 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 its table's processed retention (InboxProcessedRetentionDays / OutboxProcessedRetentionDays) 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 (with Key, CorrelationId, CausationId taken from the dead letter). 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, with one retention per table:

0 means the table is never cleaned; negative values fail start-up validation. Disable the whole service with enableCleanupHostedService: false or Cleanup.Enabled = false.

Upgrading from 3.0.5: ProcessedRetentionDays is removed. Configuration keys that are no longer bound are ignored silently, so rename it to InboxProcessedRetentionDays (and set OutboxProcessedRetentionDays if the outbox must still be cleaned; the new default keeps it forever).

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.

Schema changes in 3.0.6

The package ships no migrations; the consuming service adds one after upgrading:

• OutboxMessages: new columns CorrelationId, CausationId (length CorrelationIdMaxLength / CausationIdMaxLength, default 128); new indexes on Key, CorrelationId, CausationId.
• InboxMessages: new columns Key (KeyMaxLength, default 256), CorrelationId, CausationId; new indexes on Key, CorrelationId, CausationId.

The columns are filled on write (outbox from the envelope headers, inbox from the Kafka message key and headers, replay from the dead letter). Over-long values are cut to the column length; the full value stays in HeadersJson. Rows written before the upgrade keep NULL; HeadersJson is unchanged and still holds every header.

Pitfalls

• Consumed event types must implement INotification.
• Renaming or moving an event class changes its default type name; consumers then skip it as an unregistered type (since 3.0.6 it is marked processed with only a Warning log; earlier it was dead-lettered). Pin eventTypeName before refactoring contracts and watch for "Unregistered event type skipped" warnings after a deployment.
• 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.