如何为Confluent Helm Charts部署的Kafka配置安全连接

前置条件 #

• 已添加Confluent官方Helm仓库，Chart版本与目标Kafka版本匹配
• 推荐提前部署Cert-Manager实现证书自动管理，也可提前准备自签名/CA签发的证书文件

配置步骤 #

1. 开启TLS传输加密 #

修改Helm部署使用的values.yaml文件，开启Kafka Broker的TLS配置：

kafka:
  tls:
    enabled: true
    # 证书提供方，用Cert-Manager填cert-manager，手动上传证书填manual
    certProvider: "cert-manager"
    # 手动上传证书时需要提前在部署命名空间创建TLS Secret，取消下一行注释并填写Secret名称
    # secretRef: "kafka-broker-tls-secret"
    # 配置证书SAN（使用者可选名称），必须覆盖Broker所有访问入口：内部服务域名、外部访问域名/IP（如果有公网访问需求）
    san:
      - "*.kafka.confluent.svc.cluster.local"
      - "kafka.confluent.svc.cluster.local"

手动上传证书时，创建的Secret必须包含ca.crt、tls.crt、tls.key三个字段，证书和密钥对应Broker的服务域名。

2. 可选：开启身份认证 #

如果需要对客户端做身份校验，可选择开启双向TLS（mTLS）或SASL认证：

开启mTLS双向认证

在上述TLS配置基础上追加：

kafka:
  tls:
    authentication:
      type: "tls"

开启后客户端必须提供由同一CA签发的有效证书才能连接Broker。

开启SASL认证

配置示例如下：

kafka:
  auth:
    sasl:
      enabled: true
      # 支持的认证机制，可多选
      mechanisms: ["SCRAM-SHA-256", "SCRAM-SHA-512", "PLAIN"]
      # 配置初始管理员账号密码
      jaas:
        adminUser: "kafka-admin"
        adminPassword: "自定义强密码"

3. 部署/更新Kafka集群 #

执行Helm命令生效配置：

helm upgrade --install kafka confluentinc/cp-helm-charts -f values.yaml -n <部署命名空间>

4. 客户端连接配置 #

同K8s集群内客户端

1. 先导出集群CA证书，存到客户端所在节点：

   kubectl get secret <kafka-tls-secret名称> -n <部署命名空间> -o jsonpath='{.data.ca\.crt}' | base64 -d > ca.crt
   

2. 根据认证方式配置客户端参数：
   • 仅TLS加密无身份认证：

     bootstrap.servers=kafka.<部署命名空间>.svc.cluster.local:9093
     security.protocol=SSL
     ssl.truststore.location=/path/to/ca.crt
     

   • mTLS双向认证：
     需提前为客户端签发同一CA签名的证书密钥，追加以下配置：

     ssl.keystore.location=/path/to/client-cert.p12
     ssl.keystore.password=客户端证书密码
     

   • SASL认证：

     bootstrap.servers=kafka.<部署命名空间>.svc.cluster.local:9093
     security.protocol=SASL_SSL
     sasl.mechanism=SCRAM-SHA-256
     sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="kafka-admin" password="自定义强密码";
     ssl.truststore.location=/path/to/ca.crt
     

集群外客户端

需要额外在values.yaml中配置Kafka的外部访问规则（NodePort/LoadBalancer/Ingress），同时保证证书SAN列表包含外部访问的域名/IP，其余配置和集群内客户端一致。

连通性验证 #

可以使用Kafka官方自带的console producer/consumer工具测试连接，没有报错且能正常生产消费消息即配置成功。

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