vigilcare-clinical/docs/guides/06-apache-kafka-event-streaming.md

Guide 6: Apache Kafka Event Streaming

What is Kafka?

Apache Kafka is a distributed event streaming platform. At its simplest, it's a highly reliable message bus: producers send messages to Kafka, and consumers read them. But unlike a simple message queue, Kafka stores messages durably (on disk) and lets multiple independent consumers each read the same messages at their own pace.

Key concepts:

• Topic: A named channel for messages, like a mailbox or category. For example, observation.recorded is a topic for observation events. Producers publish to topics; consumers subscribe to topics.
• Partition: Each topic is divided into partitions (like lanes on a highway). Partitions allow parallelism — multiple consumers can read from different partitions simultaneously. Messages with the same partition key (like an encounter ID) always land on the same partition, which guarantees ordering for that key.
• Consumer Group: A named group of consumers that share the work of reading a topic. Kafka assigns each partition to exactly one consumer in the group, so messages are processed once per group. Different groups process messages independently — if both the "scoring engine" group and the "search indexer" group subscribe to the same topic, each group gets every message.
• Offset: A sequential number that identifies each message's position within a partition. Consumers track their offset (how far they've read). If a consumer crashes and restarts, it picks up where it left off.
• Producer: Code that sends messages to Kafka topics.
• Consumer: Code that reads messages from Kafka topics.

How is Kafka different from a regular queue (like RabbitMQ)? In a traditional queue, once a message is consumed, it's gone. In Kafka, messages persist (for days or longer), and multiple consumer groups can independently read the same messages. This makes Kafka ideal for event-driven architectures where one event needs to trigger many independent downstream processes.

---

Why Kafka in This Project?

When a nurse records a vital sign, the system must simultaneously: evaluate threshold breaches, compute NEWS2/qSOFA/SOFA scores, detect trends, index the observation in Elasticsearch, write it to the data lake, and potentially page a physician. Doing all of this synchronously in the HTTP request would take too long and couple unrelated systems.

Kafka decouples the write path from the processing path. The API writes an event to the outbox, the outbox relay publishes it to Kafka, and 9 independent consumer groups each process it at their own pace. If the trend analyzer is slow, the qSOFA evaluator is unaffected.

---

Architecture Overview

HTTP Request
     │
     ▼
┌──────────────┐    outbox     ┌──────────────┐
│ Observation  │───────────────│ OutboxRelay  │
│ Service      │  (PostgreSQL) │ Service      │
└──────────────┘               └──────┬───────┘
                                      │ Kafka Producer
                                      ▼
                          ┌───────────────────────┐
                          │   Kafka Broker (KRaft) │
                          │                       │
                          │  observation.recorded  │ ← 6 partitions
                          │  alert.generated       │ ← 6 partitions
                          │  encounter.status.*    │ ← 6 partitions
                          │  gcs.scored            │ ← 6 partitions
                          │  sepsis.bundle.*       │ ← 6 partitions
                          └───────┬───────────────┘
                                  │
              ┌───────────────────┼───────────────────┐
              │                   │                   │
    ┌─────────┴──────┐  ┌────────┴───────┐  ┌───────┴────────┐
    │ es-indexer      │  │ sepsis-engine  │  │ news2-scoring  │
    │ (CQRS proj.)   │  │ (qSOFA eval)   │  │ (7-param agg)  │
    ├─────────────────┤  ├────────────────┤  ├────────────────┤
    │ warning-eval.   │  │ gcs-scoring    │  │ sofa-scoring   │
    ├─────────────────┤  ├────────────────┤  ├────────────────┤
    │ trend-analyzer  │  │ notification-  │  │ data-lake-     │
    │ (rate-of-change)│  │ publisher      │  │ writer         │
    └─────────────────┘  └────────────────┘  └────────────────┘

---

Kafka Configuration

Docker Compose — KRaft Mode (No Zookeeper)

kafka:
  image: apache/kafka:3.7.0
  environment:
    KAFKA_NODE_ID: 1
    KAFKA_PROCESS_ROLES: broker,controller
    KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
    KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
    KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
    KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false"
    CLUSTER_ID: "MkU3OEVBNTcwNTJENDM2Qk"

Key decisions:

• KRaft mode (KAFKA_PROCESS_ROLES: broker,controller): Kafka 3.7 runs its own metadata consensus without Zookeeper, eliminating an entire service
• Auto-creation disabled: Topics are provisioned explicitly with 6 partitions. Auto-creation would silently create single-partition topics if the relay publishes before the provisioner runs
• Fixed CLUSTER_ID: Prevents storage reinitialization on container restart

Application Configuration

public class KafkaOptions
{
    public const string Section = "Kafka";
    public string BootstrapServers { get; set; } = null!;
    public KafkaTopicOptions Topics { get; set; } = null!;
    public int NumPartitions { get; set; } = 6;
    public short ReplicationFactor { get; set; } = 3;
    public int OutboxBatchSize { get; set; } = 100;
    public int OutboxPollIntervalMs { get; set; } = 500;
    public int OutboxMaxRetries { get; set; } = 10;
    public int MaxPoisonRetries { get; set; } = 5;
}

public class KafkaTopicOptions
{
    public string ObservationRecorded { get; set; } = "observation.recorded";
    public string AlertGenerated { get; set; } = "alert.generated";
    public string AlertAcknowledged { get; set; } = "alert.acknowledged";
    public string EncounterStatusChanged { get; set; } = "encounter.status.changed";
    public string SepsisBundleCreated { get; set; } = "sepsis.bundle.created";
    public string SepsisBundleUpdated { get; set; } = "sepsis.bundle.updated";
    public string GcsScored { get; set; } = "gcs.scored";
}

{
  "Kafka": {
    "BootstrapServers": "localhost:9092",
    "Topics": {
      "ObservationRecorded": "observation.recorded",
      "AlertGenerated": "alert.generated",
      "EncounterStatusChanged": "encounter.status.changed",
      "GcsScored": "gcs.scored"
    },
    "NumPartitions": 6,
    "OutboxBatchSize": 100,
    "OutboxPollIntervalMs": 1000
  }
}

---

Topic Provisioning

Topics are created on application startup by KafkaTopicProvisioner:

public class KafkaTopicProvisioner : IHostedService
{
    public async Task StartAsync(CancellationToken cancellationToken)
    {
        using var admin = new AdminClientBuilder(new AdminClientConfig
        {
            BootstrapServers = _options.BootstrapServers
        }).Build();

        var topicNames = new[]
        {
            _options.Topics.ObservationRecorded,
            _options.Topics.AlertGenerated,
            _options.Topics.AlertAcknowledged,
            _options.Topics.EncounterStatusChanged,
            _options.Topics.SepsisBundleCreated,
            _options.Topics.SepsisBundleUpdated,
            _options.Topics.GcsScored
        };

        var specs = topicNames.Select(name => new TopicSpecification
        {
            Name = name,
            NumPartitions = _options.NumPartitions,   // 6
            ReplicationFactor = _options.ReplicationFactor
        }).ToList();

        try
        {
            await admin.CreateTopicsAsync(specs);
            _logger.LogInformation("Kafka topics provisioned: {Topics}",
                string.Join(", ", topicNames));
        }
        catch (CreateTopicsException ex)
        {
            // TopicAlreadyExists is not an error — idempotent startup
            var errors = ex.Results
                .Where(r => r.Error.Code is not (ErrorCode.NoError or ErrorCode.TopicAlreadyExists))
                .ToList();
            if (errors.Count > 0)
                throw new InvalidOperationException(
                    $"Failed to create Kafka topics: {string.Join(", ", errors.Select(e => e.Error.Reason))}");

            _logger.LogInformation("Kafka topics already exist — skipping creation");
        }
    }
}

7 topics, all with 6 partitions. The TopicAlreadyExists error is swallowed — the provisioner is idempotent across restarts.

---

The Outbox Relay — Producing Messages

Messages are never published directly to Kafka from request handlers. Why? Because you'd face an impossible consistency problem: if you save to the database AND publish to Kafka in the same request, one might succeed and the other might fail, leaving your system in an inconsistent state. Instead, this project uses the outbox pattern (covered in detail in Guide 10): an OutboxEvent row is written to PostgreSQL in the same transaction as the domain entity (guaranteed atomic), and a background service called OutboxRelayService reads those rows and publishes them to Kafka separately.

Idempotent Producer

What does "idempotent" mean? An operation is idempotent if doing it multiple times produces the same result as doing it once. An idempotent Kafka producer ensures that if a message is accidentally sent twice (due to a network timeout and retry), Kafka stores it only once:

_producer = new ProducerBuilder<string, string>(new ProducerConfig
{
    BootstrapServers = _options.BootstrapServers,
    Acks = Acks.All,
    EnableIdempotence = true,
    MessageSendMaxRetries = 3,
    RetryBackoffMs = 100
}).Build();

• Acks.All: After the producer sends a message, it waits for confirmation. All means the Kafka broker confirms only after all replica copies have stored the message. This is the highest durability guarantee — your message won't be lost even if a broker crashes.
• EnableIdempotence = true: The broker assigns each producer an internal ID and sequence number. If a network timeout causes the client to retry a message the broker already accepted, the broker recognizes the duplicate by its sequence number and silently drops it.

Publishing Logic

private async Task ProcessBatchAsync(CancellationToken ct)
{
    var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
    await using var tx = await db.Database.BeginTransactionAsync(ct);

    // Lock rows, skip any already locked by another relay instance
    var events = await db.OutboxEvents
        .FromSqlRaw("""
            SELECT ... FROM outbox_events
            WHERE processed_at IS NULL AND failed_at IS NULL
            ORDER BY created_at ASC
            LIMIT {0}
            FOR UPDATE SKIP LOCKED
            """, _options.OutboxBatchSize)
        .ToListAsync(ct);

    foreach (var ev in events)
    {
        try
        {
            var result = await _producer!.ProduceAsync(
                ev.Topic,
                new Message<string, string>
                {
                    Key = ev.PartitionKey ?? string.Empty,
                    Value = ev.Payload
                }, ct);

            ev.ProcessedAt = DateTimeOffset.UtcNow;
        }
        catch (ProduceException<string, string> ex)
        {
            ev.RetryCount++;
            ev.LastError = ex.Error.Reason;

            if (ev.RetryCount >= _options.OutboxMaxRetries)
            {
                ev.FailedAt = DateTimeOffset.UtcNow;
                _logger.LogError(ex,
                    "Outbox event {Id} permanently failed after {Retries} retries",
                    ev.Id, ev.RetryCount);
            }
            break;  // Stop processing batch on first failure
        }
    }

    await db.SaveChangesAsync(ct);
    await tx.CommitAsync(ct);
}

Key patterns:

• Partition key is the encounterId — all events for one encounter land on the same partition, guaranteeing ordering per encounter
• Retry tracking — each failed event increments RetryCount and records LastError. After OutboxMaxRetries (10), the event is marked as permanently failed with FailedAt
• Break on failure — if one event fails to publish, the batch stops. This prevents out-of-order delivery within an encounter.

---

Consumer Patterns

All consumers follow the same structural pattern with minor variations. Understanding this pattern is important because you'll see it repeated across 9 different services.

Pattern 1: Simple Consumer (SepsisEngineService)

The simplest pattern — subscribe to one topic, read messages in a loop, process each one, and commit the offset (tell Kafka "I've finished processing this message"):

protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
    var config = new ConsumerConfig
    {
        BootstrapServers = _kafkaOptions.BootstrapServers,
        GroupId = "sepsis-engine",
        AutoOffsetReset = AutoOffsetReset.Earliest,
        EnableAutoCommit = false
    };

    using var consumer = new ConsumerBuilder<string, string>(config).Build();
    consumer.Subscribe(_kafkaOptions.Topics.ObservationRecorded);

    var guard = new PoisonPillGuard("sepsis-engine", _kafkaOptions.MaxPoisonRetries, _logger);

    while (!stoppingToken.IsCancellationRequested)
    {
        ConsumeResult<string, string>? result = null;
        try
        {
            result = consumer.Consume(stoppingToken);

            var evt = JsonSerializer.Deserialize<SepsisObservationEvent>(
                result.Message.Value)!;

            using var scope = _services.CreateScope();
            var qsofaDetector = scope.ServiceProvider.GetRequiredService<QsofaDetector>();
            await qsofaDetector.ProcessObservationAsync(
                evt.EncounterId, evt.PatientId, evt.ObservationCode, evt.Value, stoppingToken);

            consumer.Commit(result);
            guard.OnSuccess();
        }
        catch (OperationCanceledException) { break; }
        catch (Exception ex)
        {
            if (result is not null && guard.ShouldSkip(result, ex))
            {
                consumer.Commit(result);
                continue;
            }
            _logger.LogError(ex, "SepsisEngine failed — will retry");
            await Task.Delay(2000, stoppingToken);
        }
    }
    consumer.Close();
}

Common configuration across all consumers:

• AutoOffsetReset = Earliest: When a consumer starts for the first time (no previously committed offset), should it read from the beginning of the topic (Earliest) or only new messages (Latest)? We use Earliest to ensure no events are missed — even if the consumer starts hours after the topic was created.
• EnableAutoCommit = false: By default, Kafka consumers automatically commit their offset every 5 seconds (telling Kafka "I've processed everything up to here"). But what if the consumer crashes between the auto-commit and actually finishing the work? The message would be marked as processed but never actually handled. Manual commit means we only tell Kafka "done" after we've confirmed the work succeeded. This gives us at-least-once delivery — a message might be processed twice (if the consumer crashes after processing but before committing), but it will never be lost.
• consumer.Close() in the finally block: Tells the Kafka broker "I'm leaving the group." This triggers an immediate rebalance so other consumers pick up this consumer's partitions right away, instead of waiting for a session timeout (typically 30 seconds).

Pattern 2: Multi-Topic Consumer (NotificationPublisherService)

Subscribes to multiple topics and routes based on the topic name:

consumer.Subscribe(new[]
{
    _kafkaOptions.Topics.AlertGenerated,
    _kafkaOptions.Topics.EncounterStatusChanged,
});

// In the consume loop:
if (result.Topic == _kafkaOptions.Topics.AlertGenerated)
    await HandleAlertGeneratedAsync(chan, props, result.Message.Value, stoppingToken);
else if (result.Topic == _kafkaOptions.Topics.EncounterStatusChanged)
    await HandleEncounterStatusChangedAsync(chan, props, result.Message.Value, stoppingToken);

This consumer bridges Kafka → RabbitMQ: critical alerts go to the paging queue, discharged encounters go to the discharge summary queue.

Pattern 3: Buffered Consumer (DataLakeWriterService)

Buffers events in memory and flushes to MinIO in batches:

consumer.Subscribe(new[]
{
    _kafkaOptions.Topics.ObservationRecorded,
    _kafkaOptions.Topics.AlertGenerated,
    _kafkaOptions.Topics.EncounterStatusChanged,
});

// Non-blocking consume with timeout
result = consumer.Consume(TimeSpan.FromMilliseconds(500));

if (result is not null)
    AddToBuffer(result);

// Flush when buffer is full or timer expires
var shouldFlushCount = totalBuffered >= _opts.FlushCount;    // 1000 events
var shouldFlushTime  = elapsed >= TimeSpan.FromSeconds(_opts.FlushIntervalSeconds);  // 300s

The data lake writer tracks per-partition high watermarks and only commits offsets for partitions where all MinIO uploads succeeded — partial-commit safety.

---

The Poison Pill Guard

What is a poison pill? In messaging systems, a "poison pill" is a message that a consumer cannot process — maybe the JSON is malformed, a required field is missing, or the data violates a business rule. Without protection, the consumer reads the message, fails to process it, doesn't commit the offset, and reads the same message again on the next loop iteration — forever. The consumer is stuck in an infinite retry loop, and all subsequent messages on that partition are blocked behind it.

The PoisonPillGuard detects this situation and skips unprocessable messages:

public sealed class PoisonPillGuard
{
    private static readonly Counter PoisonPillsSkipped = Metrics.CreateCounter(
        "kafka_poison_pills_skipped_total",
        "Messages skipped as poison pills.",
        labelNames: new[] { "consumer_group", "topic" });

    public bool ShouldSkip(ConsumeResult<string, string> result, Exception ex)
    {
        // Permanent errors — skip immediately
        if (IsPermanent(ex))
        {
            LogSkip(result, ex, "permanent");
            return true;
        }

        // Transient errors — retry up to maxRetries
        var key = (result.Topic, result.Partition.Value, result.Offset.Value);
        if (_lastFailedKey == key)
            _retryCount++;
        else
        {
            _lastFailedKey = key;
            _retryCount = 1;
        }

        if (_retryCount >= _maxRetries)
        {
            LogSkip(result, ex, $"transient after {_retryCount} retries");
            return true;
        }

        return false;
    }

    public void OnSuccess()
    {
        _lastFailedKey = null;
        _retryCount = 0;
    }

    private static bool IsPermanent(Exception ex) =>
        GetRoot(ex) is JsonException or FormatException or ArgumentNullException;
}

Classification:

• Permanent errors (JsonException, FormatException, ArgumentNullException): The message payload is malformed — retrying won't help. Skip immediately.
• Transient errors (database timeout, Redis unavailable): Retry up to MaxPoisonRetries (5). If still failing, skip and move on.

Skipped messages are:

1. Logged at Critical level with the full payload (truncated to 2000 chars)
2. Counted in the kafka_poison_pills_skipped_total Prometheus metric
3. Committed (offset advances past the poison pill)

---

Consumer Group Summary

Group ID	Topics	Purpose	Service
es-indexer	observation.recorded, alert.generated, encounter.status.changed, sepsis.bundle.*, gcs.scored	CQRS projection to Elasticsearch	EsIndexerService
sepsis-engine	observation.recorded	qSOFA screening	SepsisEngineService
warning-evaluator	observation.recorded	Warning-range threshold alerts	WarningAlertService
news2-scoring	observation.recorded	NEWS2 composite score	News2ScoringService
gcs-scoring	observation.recorded	GCS component aggregation	GcsScoringService
sofa-scoring	observation.recorded, gcs.scored	SOFA organ-dysfunction scoring	SofaScoringService
trend-analyzer	observation.recorded	Rate-of-change detection	TrendAnalyzerService
notification-publisher	alert.generated, encounter.status.changed	Kafka → RabbitMQ bridge	NotificationPublisherService
data-lake-writer	observation.recorded, alert.generated, encounter.status.changed	Parquet files to MinIO	DataLakeWriterService

All 9 consumer groups process observation.recorded independently. Publishing one observation event triggers 7+ parallel processing paths.

---

Topic Summary

Topic	Partition Key	Producers	Consumers
observation.recorded	encounterId	OutboxRelay	es-indexer, sepsis-engine, warning-evaluator, news2-scoring, gcs-scoring, sofa-scoring, trend-analyzer, data-lake-writer
alert.generated	encounterId	OutboxRelay	es-indexer, notification-publisher, data-lake-writer
alert.acknowledged	encounterId	OutboxRelay	es-indexer
encounter.status.changed	encounterId	OutboxRelay	es-indexer, notification-publisher, data-lake-writer
gcs.scored	encounterId	GcsScoringService (direct)	sofa-scoring, es-indexer
sepsis.bundle.created	encounterId	OutboxRelay	es-indexer
sepsis.bundle.updated	encounterId	OutboxRelay	es-indexer

All topics use encounterId as the partition key. This guarantees that all events for a single encounter are processed in order within each consumer group, which is essential for clinical correctness (you can't evaluate qSOFA before the observation that triggered it).

---

Health Check

public sealed class KafkaHealthCheck : IHealthCheck
{
    public async Task<HealthCheckResult> CheckHealthAsync(
        HealthCheckContext context, CancellationToken cancellationToken)
    {
        using var admin = new AdminClientBuilder(new AdminClientConfig
        {
            BootstrapServers = _options.BootstrapServers
        }).Build();

        var metadata = await Task.Run(
            () => admin.GetMetadata(TimeSpan.FromSeconds(5)), cancellationToken);

        var data = new Dictionary<string, object> { ["brokers"] = metadata.Brokers.Count };
        return HealthCheckResult.Healthy(data: data);
    }
}

Queries the broker metadata to verify the cluster is reachable and has at least one broker.

---

Monitoring

Consumer Lag Collector

The KafkaConsumerLagCollector polls offset lag every 30 seconds for 4 key consumer groups:

private static readonly string[] Groups =
{
    "es-indexer",
    "sepsis-engine",
    "notification-publisher",
    "data-lake-writer",
};

For each group:

1. Query committed offsets via the Admin API
2. Query high watermarks via a temporary consumer
3. Lag = high watermark − committed offset (summed across all partitions)
4. Set kafka_consumer_lag gauge with the group ID as a label

A rising lag means a consumer is falling behind the event stream — visible instantly on the Grafana dashboard.

---

Key Design Decisions

Why Outbox + Relay Instead of Direct Publish?

Imagine your code does this: (1) save observation to database, (2) publish event to Kafka. What happens if step 1 succeeds but step 2 fails (network blip)? You have data in your database that downstream consumers never learn about. What if step 2 succeeds but step 1 fails? Consumers process an event for data that doesn't exist.

The outbox pattern solves this by writing BOTH the domain entity and the outbox event in one PostgreSQL transaction — they either both succeed or both fail. The relay service reads the outbox table separately and publishes to Kafka. If the relay fails, it retries later (the outbox row is still there). The idempotent producer prevents duplicates from retries. This gives you guaranteed eventual delivery with no data loss.

Why 6 Partitions?

In Kafka, the number of partitions determines the maximum parallelism for a consumer group — one consumer instance can read from one partition. With 6 partitions, you can run up to 6 consumer instances per group. For local development with a single instance, that instance reads all 6 partitions. In production, you could scale to 6 instances per consumer group for horizontal parallelism. The partition key (encounterId) distributes encounters evenly across partitions via a hash function.

Why Manual Commit?

As explained in the consumer configuration section, auto-commit periodically tells Kafka "I've processed everything" — even if you haven't. This means messages can be lost if the consumer crashes at the wrong moment. Manual commit after successful processing guarantees at-least-once delivery: a message might be processed twice (rare, only on crash), but it will never be silently lost. For a patient safety system, losing a vital sign observation is unacceptable, so at-least-once is the right tradeoff.