# DFE on AWS MSK

> Operations (bootstrap Job, autoscaling, telemetry):
> [aws-msk-operations.md](/v/2.2.1/dfe-infra/deployment/kafka/aws-msk-operations.md).

The aws root builds MSK Provisioned itself under `kafka.provider: msk`
(`terraform/modules/managed-kafka/msk`). This guide is for an MSK you stand up
yourself, then point DFE at with `kafka.mode=external`. MSK comes in two shapes
DFE supports as distinct providers:

- **MSK Provisioned** -- fixed broker-hour capacity, KRaft, full
  `server.properties` control. DFE provider `msk`, auth **SASL/SCRAM-SHA-512**
  (Secrets Manager). This is the closest managed cluster to DFE's own
  self-hosted brokers.
- **MSK Serverless** -- elastic, IAM-only. DFE provider `msk_iam`, auth
  **IAM / SASL_SSL OAUTHBEARER** (port 9098). No static secret; almost no broker
  config.

Read the opinionated baseline in [README.md](/v/2.2.1/dfe-infra/deployment/kafka.md) first -- this guide only
covers what MSK lets you apply and how.

MSK Provisioned is the slowest backend to create and delete, and a cluster can't be paused, so it bills for every hour it exists. Plan any test environment around that. DFE was tested on MSK once, with 2.2.1, and is not re-tested there without explicit approval: only where a change needs a real MSK cluster to prove, or alongside AWS if they want to be involved ([TESTING-CYCLE.md](https://github.com/hyperi-io/dfe-infra/blob/2.2.1/docs/TESTING-CYCLE.md#msk-runs-last-and-rarely)).

## What MSK lets you set vs locks down

### MSK Provisioned

MSK Provisioned gives you a real `server.properties` via an
`aws_msk_configuration` attached to the cluster, so the broker-level DFE defaults
apply directly.

| DFE default | On MSK Provisioned |
|---|---|
| KRaft (no ZooKeeper) | Yes -- MSK is KRaft-managed; nothing to set. |
| `default.replication.factor=3` | Settable, and MSK FORCES topic RF to 3 anyway (a create with RF != 3 is coerced to 3). Aligned with DFE. |
| `min.insync.replicas=2` | Settable at the broker level (`server_properties`). NOTE: the topic-level override is PROTECTED -- you cannot weaken it per topic. |
| `unclean.leader.election.enable=false` | Settable at the broker level. Topic-level override PROTECTED. |
| `num.partitions=12` | Settable at the broker level (applies to auto-created topics). |
| `compression.type=producer` | Settable. |
| `log.retention.hours=72` | Settable. |
| `log.retention.bytes` (unset) | Settable (leave off / -1). |
| `log.segment.bytes=512 MiB` | Settable. |
| `message.max.bytes` (default) | Settable; keep aligned with `replica.fetch.max.bytes`. |

Locked: the topic-level overrides of `min.insync.replicas`,
`unclean.leader.election.enable`, and the replication factor are protected by
MSK. This is FINE for DFE -- the DFE defaults are exactly what MSK enforces, so
the protection works in your favour.

### MSK Serverless

MSK Serverless hides the broker entirely. There is **no `aws_msk_configuration`**
-- you cannot set `num.partitions`, `default.replication.factor`,
`unclean.leader.election.enable`, `log.segment.bytes`, or any other broker
default. AWS manages durability (each partition gets replicas AWS does not
count against your quota).

| DFE default | On MSK Serverless |
|---|---|
| KRaft / replication / ISR / unclean election | Locked -- AWS-managed, not exposed. Durability is handled by AWS; you cannot set RF/ISR. |
| `num.partitions=12` | Not a broker default -- set partition count PER TOPIC at create time. Cluster quota: 2400 leader partitions (raise via AWS Support). |
| `compression.type`, `retention.ms`, `retention.bytes`, `max.message.bytes`, `cleanup.policy` | Settable PER TOPIC only (these five are the honoured topic configs). |
| `log.segment.bytes` | Locked -- managed. |
| Per-partition storage | Hard cap 250 GB/partition -- add partitions for more storage, not longer single-partition retention. |

Because Serverless exposes so little, the DFE opinion collapses to: RF/ISR/unclean
are AWS-managed (already durable), and you express retention + compression +
partitions per topic.

## Terraform / OpenTofu -- MSK Provisioned (SASL/SCRAM-512)

This is the DFE `msk` provider path: a provisioned cluster with a custom
configuration carrying the DFE broker defaults, SASL/SCRAM auth backed by Secrets
Manager, and the SCRAM secret associated to the cluster. The Kafka version,
broker count, broker type and volume size are variables your tfvars set, never
literals in the `.tf`.

```hcl
terraform {
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "~> 6.0" # VERIFY latest + PIN exact (6.54.0 seen 2026-07-08); see README pinning note.
    }
  }
}

# --- Custom broker configuration: the DFE opinionated defaults ---
resource "aws_msk_configuration" "dfe" {
  kafka_versions = [var.kafka_version] # aws kafka list-kafka-versions prints the ACTIVE ones
  name           = "dfe-kafka"

  server_properties = <<-PROPERTIES
    auto.create.topics.enable=false
    default.replication.factor=3
    min.insync.replicas=2
    unclean.leader.election.enable=false
    num.partitions=12
    compression.type=producer
    log.retention.hours=72
    log.segment.bytes=536870912
    replica.lag.time.max.ms=30000
  PROPERTIES
}

# --- KMS key for the SCRAM secret (Secrets Manager requires a CMK, not the AWS-managed key) ---
resource "aws_kms_key" "dfe_kafka" {
  description = "DFE MSK SCRAM credential encryption"
}

# --- SCRAM credential in Secrets Manager (name MUST start with "AmazonMSK_") ---
resource "aws_secretsmanager_secret" "dfe_kafka_user" {
  name       = "AmazonMSK_dfe-kafka-user"
  kms_key_id = aws_kms_key.dfe_kafka.key_id
}

resource "aws_secretsmanager_secret_version" "dfe_kafka_user" {
  secret_id     = aws_secretsmanager_secret.dfe_kafka_user.id
  secret_string = jsonencode({ username = "dfe-kafka-user", password = var.dfe_kafka_password })
}

# --- The provisioned cluster ---
resource "aws_msk_cluster" "dfe" {
  cluster_name           = "dfe-kafka"
  kafka_version          = var.kafka_version
  number_of_broker_nodes = var.broker_count # a multiple of the subnet count, RF3 across AZs

  broker_node_group_info {
    instance_type   = var.broker_instance_type
    client_subnets  = var.private_subnet_ids # 3 subnets in 3 AZs
    security_groups = [aws_security_group.msk.id]
    storage_info {
      ebs_storage_info { volume_size = var.broker_volume_gib }
    }
  }

  configuration_info {
    arn      = aws_msk_configuration.dfe.arn
    revision = aws_msk_configuration.dfe.latest_revision
  }

  client_authentication {
    sasl { scram = true }
  }

  encryption_info {
    encryption_in_transit {
      client_broker = "TLS"        # SASL_SSL: SCRAM over TLS -- the DFE floor
      in_cluster    = true
    }
  }
}

# --- Bind the SCRAM secret to the cluster ---
resource "aws_msk_scram_secret_association" "dfe" {
  cluster_arn     = aws_msk_cluster.dfe.arn
  secret_arn_list = [aws_secretsmanager_secret.dfe_kafka_user.arn]
  depends_on      = [aws_secretsmanager_secret_version.dfe_kafka_user]
}

output "bootstrap_brokers_sasl_scram" {
  value = aws_msk_cluster.dfe.bootstrap_brokers_sasl_scram
}
```

Point DFE at the `bootstrap_brokers_sasl_scram` endpoint with provider `msk`
(see "Wiring back to DFE" below).

## Terraform / OpenTofu -- MSK Serverless (IAM)

This is the DFE `msk_iam` provider path. Minimal cluster (AWS manages capacity);
auth is IAM only, so there is NO Secrets Manager secret and NO
`aws_msk_configuration`.

```hcl
resource "aws_msk_serverless_cluster" "dfe" {
  cluster_name = "dfe-kafka-serverless"

  vpc_config {
    subnet_ids         = var.private_subnet_ids
    security_group_ids = [aws_security_group.msk.id]
  }

  client_authentication {
    sasl { iam { enabled = true } } # IAM only; port 9098
  }
}

# IAM policy the DFE consumer/producer role needs (attach to the IRSA role that
# the DFE ServiceAccounts assume -- there is no static credential).
data "aws_iam_policy_document" "dfe_kafka" {
  statement {
    actions   = ["kafka-cluster:Connect", "kafka-cluster:DescribeCluster"]
    resources = [aws_msk_serverless_cluster.dfe.arn]
  }
  statement {
    actions = [
      "kafka-cluster:CreateTopic", "kafka-cluster:DescribeTopic",
      "kafka-cluster:WriteData", "kafka-cluster:ReadData",
    ]
    resources = ["${replace(aws_msk_serverless_cluster.dfe.arn, ":cluster/", ":topic/")}/*"]
  }
  statement {
    actions   = ["kafka-cluster:AlterGroup", "kafka-cluster:DescribeGroup"]
    resources = ["${replace(aws_msk_serverless_cluster.dfe.arn, ":cluster/", ":group/")}/dfe-*"]
  }
}

output "bootstrap_brokers_sasl_iam" {
  value = aws_msk_serverless_cluster.dfe.bootstrap_brokers_sasl_iam
}
```

### Topics on either MSK (the AWS provider does not manage topics)

Neither `aws_msk_cluster` nor `aws_msk_serverless_cluster` creates topics, and on
a Provisioned cluster carrying the DFE configuration nothing else will either:
`auto.create.topics.enable` is false, so that the partition count per source is an
explicit number rather than a broker default. dfe-engine creates them at startup
from the topic set dfe-schemas declares, under the ACLs the bootstrap Job below
writes. Leave them to it: a topic created any other way keeps its own
partitions and configs, which the engine does not reshape.

See [aws-msk-operations.md](/v/2.2.1/dfe-infra/deployment/kafka/aws-msk-operations.md) for the bootstrap Job
that writes the ACLs (and what you set to drive it by hand),
broker-count autoscaling, telemetry routing, and the broker security
group's ports.

## Wiring back to DFE

- `kafka.mode=external`, `kafka.external.bootstrap=<bootstrap endpoint>`.
- `kafka.securityProtocol=SASL_SSL`: the SCRAM listener on 9096 is TLS only, and every app chart and dfe-engine read their protocol from this one dial. On the aws root you set none of this by hand: `DFE_KAFKA_PROVIDER=msk` reaches bootstrap, which puts the mode, the endpoint and `SASL_SSL` on the Argo cluster secret for the layer 2 appsets to hand on.
- `kafka.external.provider`, the identity the chart derives the SASL mechanism from:
  - Provisioned: `msk` -> `SASL_SSL` + `SCRAM-SHA-512`. Credentials via the Vault-backed external secret (`kafka.external.auth.type=scram`).
  - Serverless: `msk_iam` -> `SASL_SSL` + `OAUTHBEARER`. Set `kafka.external.auth.type=msk_iam` -- NO static secret is rendered.
- Client config: `dfe kafka client-config --provider msk` (or `msk_iam`) emits
  the derived settings from the canonical provider table.

## Verification -- what is and is not confirmed

Confirmed by 2026 research (sources in README): MSK Provisioned exposes
`server_properties` including all the DFE broker defaults; topic-level RF /
`min.insync.replicas` / `unclean.leader.election.enable` are protected and RF is
forced to 3. MSK Serverless is IAM-only on 9098, has no `aws_msk_configuration`,
caps at 2400 leader partitions and 250 GB/partition, and honours only
`cleanup.policy`, `compression.type`, `max.message.bytes`, `retention.bytes`,
`retention.ms` per topic. The `aws_msk_configuration`,
`aws_msk_scram_secret_association`, `aws_msk_serverless_cluster`, and IAM SASL
resources are all current in the provider.

The bootstrap Job's pinned jar is verified: `aws-msk-iam-auth` v2.3.8 was
published 2026-09-05 and the sha256 in `values.yaml` matches both the release's
own `.sha256` asset and a hand-computed sum of the downloaded jar.

NOT verified (no live AWS account in this workstream, cannot deploy): that the
bootstrap Job authenticates end to end -- the Pod Identity credential reaching the
IAM login module, and `kafka-acls.sh` writing an ACL over SASL/IAM, have not been
run against a real cluster. Also the
`broker_instance_type` / `broker_volume_gib` sizing for your throughput, and that every `server.properties` key above is permitted
on the specific Kafka version you pick (the allowed set varies by version --
check the AWS "Custom MSK configurations" page). Treat the snippets as templates
to `tofu validate` and plan against your account before apply.