DFE on AWS MSK

Operations (bootstrap Job, autoscaling, telemetry): 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 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).

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.

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.

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