# 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)

```yaml
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

```csharp
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";
}
```

```json
{
  "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`:

```csharp
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:

```csharp
_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

```csharp
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"):

```csharp
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:

```csharp
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:

```csharp
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:

```csharp
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

```csharp
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:

```csharp
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.