同VPC下无公网IP的Dataproc集群无需NAT访问GKE Kafka方案咨询

问题根源 #

当前依赖NAT通过公网IP访问的核心原因是：Kafka Broker配置的advertised.listeners为公网地址。Spark客户端连接Kafka时，会收到Broker返回的公网连接地址，而私有Dataproc集群无公网IP，没有NAT就无法解析并连接该地址。

具体解决方案 #

1. 修改Kafka监听地址配置 #

将Kafka Broker的advertised.listeners改为VPC内网地址或GKE集群内部DNS名称，确保Spark客户端收到的是可直接访问的内网连接信息：

• 若用StatefulSet部署Kafka，每个Broker的内网DNS格式为：<statefulset-name>-<index>.<service-name>.<namespace>.svc.cluster.local（例如kafka-0.kafka-service.default.svc.cluster.local）
• 在Kafka配置文件中更新：

  # 保留SSL监听配置
  listeners=SSL://0.0.0.0:9093
  # 设置为内网地址/内部DNS列表
  advertised.listeners=SSL://kafka-0.kafka-service.default.svc.cluster.local:9093,SSL://kafka-1.kafka-service.default.svc.cluster.local:9093
  ssl.keystore.location=/path/to/keystore.jks
  ssl.keystore.password=your-keystore-password
  ssl.truststore.location=/path/to/truststore.jks
  ssl.truststore.password=your-truststore-password
  

• 若使用Headless Service，可直接用Service的DNS前缀让客户端自动发现Broker，无需逐个指定IP。

2. 配置VPC防火墙规则 #

确保VPC内允许Dataproc子网的流量访问Kafka Broker的SSL端口（默认9093）：

• 创建防火墙规则，源IP范围设为Dataproc集群所在子网的CIDR（例如10.0.1.0/24），目标IP范围设为GKE集群所在子网的CIDR
• 允许的协议端口为tcp:9093

3. 调整Spark作业的Kafka连接参数 #

在Spark作业中指定Kafka的内网地址作为bootstrap服务器，并配置正确的SSL参数：

val kafkaParams = Map(
  "bootstrap.servers" -> "kafka-0.kafka-service.default.svc.cluster.local:9093,kafka-1.kafka-service.default.svc.cluster.local:9093",
  "security.protocol" -> "SSL",
  "ssl.truststore.location" -> "/usr/lib/spark/conf/truststore.jks", // 证书存放路径
  "ssl.truststore.password" -> "your-truststore-password",
  "subscribe" -> "your-topic-name"
)

val df = spark.readStream
  .format("kafka")
  .options(kafkaParams)
  .load()

• 可通过Dataproc初始化脚本将Kafka的SSL证书（truststore）分发到所有节点，或者从GCS桶加载证书文件。

4. 可选：使用GKE内部负载均衡简化访问 #

若不想维护Broker地址列表，可创建GKE内部LoadBalancer Service，让Dataproc通过固定内网IP访问Kafka集群：

apiVersion: v1
kind: Service
metadata:
  name: kafka-internal-lb
  annotations:
    cloud.google.com/load-balancer-type: "Internal" # 指定为内部负载均衡
spec:
  type: LoadBalancer
  selector:
    app: kafka # 匹配Kafka Pod的标签
  ports:
    - name: ssl
      port: 9093
      targetPort: 9093
  loadBalancerIP: 10.0.2.100 # 可选，指定VPC内预留的静态内网IP

此时Spark作业的bootstrap.servers只需设为该负载均衡的内网IP即可。

验证步骤 #

1. 在Dataproc主节点执行nc -zv <kafka-internal-ip> 9093，确认端口可连通
2. 使用Kafka命令行工具测试SSL连接：

   kafka-console-consumer.sh --bootstrap-server <kafka-internal-ip>:9093 --topic your-topic-name --from-beginning --consumer.config consumer-ssl.properties
   

   其中consumer-ssl.properties包含SSL信任库配置。

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