如何在AWS MSK中实现Kafka主题的配置即代码？是否支持类Strimzi的配置创建主题？

以下是几种可行的实现方式：

1. 结合基础设施即代码（IaC）工具 #

Terraform #

使用Terraform的aws_msk_topic资源直接在基础设施代码中定义主题的所有配置，部署时自动同步到MSK集群：

resource "aws_msk_topic" "user_activity" {
  cluster_arn = aws_msk_cluster.prod.arn
  name        = "user-activity"
  partitions  = 6
  replicas    = 3

  config {
    "cleanup.policy"      = "delete"
    "retention.ms"        = "86400000"
    "min.insync.replicas" = "2"
  }
}

这种方式的优势是主题配置与集群基础设施代码统一管理，版本控制更方便。

CloudFormation #

通过AWS::MSK::Topic资源在CloudFormation模板中声明主题，支持创建、更新和删除操作：

Resources:
  UserActivityTopic:
    Type: AWS::MSK::Topic
    Properties:
      ClusterArn: !Ref ProdMSKCluster
      TopicName: user-activity
      NumberOfPartitions: 6
      NumberOfReplicas: 3
      TopicConfig:
        cleanup.policy: delete
        retention.ms: 86400000

2. 基于Kafka原生工具的CI/CD流水线 #

编写结构化的主题配置文件（如YAML），然后在CI/CD流程中通过Kafka Admin API或命令行工具批量执行管理操作：

步骤1：定义主题配置文件 #

topics:
  - name: order-processing
    partitions: 8
    replicas: 3
    configs:
      retention.ms: 604800000
      cleanup.policy: compact
  - name: inventory-updates
    partitions: 4
    replicas: 3
    configs:
      retention.ms: 259200000

步骤2：编写自动化脚本 #

使用kafka-topics.sh或编程语言封装的Admin API实现配置同步，比如Python脚本（基于confluent-kafka）：

from confluent_kafka.admin import AdminClient, NewTopic

admin_client = AdminClient({"bootstrap.servers": "<MSK_BOOTSTRAP_SERVERS>"})

# 从配置文件加载主题定义
topics_config = [
    {"name": "order-processing", "partitions": 8, "replicas": 3, "configs": {"retention.ms": "604800000"}},
    {"name": "inventory-updates", "partitions": 4, "replicas": 3, "configs": {"retention.ms": "259200000"}}
]

new_topics = [
    NewTopic(
        topic["name"],
        num_partitions=topic["partitions"],
        replication_factor=topic["replicas"],
        config=topic["configs"]
    )
    for topic in topics_config
]

# 创建主题
fs = admin_client.create_topics(new_topics)

# 等待操作完成
for topic, f in fs.items():
    try:
        f.result()
        print(f"Topic {topic} created successfully")
    except Exception as e:
        print(f"Failed to create topic {topic}: {e}")

AWS MSK本身不原生支持Strimzi通过Kubernetes CRD声明式管理主题的模式——Strimzi是Kubernetes生态中的Kafka运营商，通过自定义资源让用户以YAML声明主题，由运营商自动处理生命周期。

不过可以通过以下方式实现类似的声明式体验：

• EKS环境下结合Strimzi：如果你的MSK集群部署在EKS中，可以部署Strimzi运营商，配置它连接到MSK集群，之后就能用Strimzi的KafkaTopic CRD来声明主题，运营商会通过MSK的API完成主题的创建和维护。
• 自定义AWS Config规则：创建AWS Config自定义规则，对比预定义的主题配置清单与MSK集群的实际主题状态，当出现差异时自动触发Lambda函数执行修正操作（创建/更新/删除主题）。

内容的提问来源于stack exchange，提问作者clearpath