# Kafka (MSK)

Provision an MSK Kafka cluster and connect it to Kloudfuse.

## Prerequisites

- Kafka 3.7.x
- Instance class: `kafka.m5.large` or larger
- SASL/SCRAM + TLS authentication
- Same VPC as the EKS cluster

Set the required environment variable before running the commands on this page:

```console
export NAMESPACE=<your-namespace>
export KAFKA_PASSWORD=<your-password>
```

## Create Kubernetes Secret

```console
kubectl create secret generic kfuse-kafka-sasl-credentials \
  --namespace="$NAMESPACE" \
  --from-literal=kafka-sasl-password="$KAFKA_PASSWORD" \
  --from-literal=password="$KAFKA_PASSWORD"
```

## Create MSK Configuration

```console
cat > /tmp/msk-config.properties << 'EOF'
auto.create.topics.enable=false
delete.topic.enable=true
log.retention.bytes=10737418240
message.max.bytes=20971520
num.io.threads=64
num.network.threads=64
num.partitions=1
num.recovery.threads.per.data.dir=8
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=2
offsets.topic.replication.factor=2
allow.everyone.if.no.acl.found=true
EOF

aws kafka create-configuration \
  --name "<cluster-name>-msk-config" \
  --kafka-versions "3.7.x" \
  --server-properties fileb:///tmp/msk-config.properties
```

## Create KMS Key

MSK requires a Customer Managed Key (CMK) for SCRAM secrets — the default `aws/secretsmanager` key is not supported.

```console
aws kms create-key --description "KMS key for MSK SCRAM secrets"
aws kms create-alias \
  --alias-name "alias/<cluster-name>-msk-scram" \
  --target-key-id <key-id-from-above>
```

## Create SCRAM Secret and Associate

The secret name **must** start with `AmazonMSK_` and must be encrypted with the CMK created above.

```console
aws secretsmanager create-secret \
  --name "AmazonMSK_<cluster-name>-scram" \
  --kms-key-id "alias/<cluster-name>-msk-scram" \
  --secret-string "{\"username\":\"kfuse-admin\",\"password\":\"$KAFKA_PASSWORD\"}"

CLUSTER_ARN=$(aws kafka list-clusters-v2 \
  --query "ClusterInfoList[?ClusterName=='<cluster-name>'].ClusterArn" --output text)
SECRET_ARN=$(aws secretsmanager describe-secret \
  --secret-id "AmazonMSK_<cluster-name>-scram" --query 'ARN' --output text)

aws kafka batch-associate-scram-secret \
  --cluster-arn "$CLUSTER_ARN" \
  --secret-arn-list "$SECRET_ARN"
```

## Get Bootstrap Servers

```console
aws kafka get-bootstrap-brokers --cluster-arn "$CLUSTER_ARN" \
  --query 'BootstrapBrokerStringSaslScram' --output text
```

Use port `9096` (SASL/SCRAM). Do not use port `9092` (plaintext) or `9098` (IAM).

## Helm Values

Add the following to your `custom-values.yaml`. Use the bootstrap server addresses retrieved in [Get Bootstrap Servers](https://docs.kloudfuse.com/platform/4.0.2/setup/cloud-services/kafka/#get-bootstrap-servers).

- Standard (kfuse)
- FED (kfuse-fed)

```yaml
global:
  kafka:
    bootstrapServers: "<broker-1>:9096,<broker-2>:9096,<broker-3>:9096"
```

```yaml
global:
  kafka:
    bootstrapServers: "<broker-1>:9096,<broker-2>:9096,<broker-3>:9096"
```