Kubernetes集群中FastAPI应用无法连接Kafka问题求助

问题背景 #

我在同一Kubernetes集群的不同namespace中部署了Python FastAPI应用与Kafka服务，FastAPI应用需向Kafka Topic发送消息，但启动时崩溃并抛出NoBrokersAvailable错误。奇怪的是，使用相同连接配置的普通Python Pod（未集成FastAPI）可以正常连接Kafka Broker。

连接Kafka的Python代码 #

producer = KafkaProducer(
    bootstrap_servers=os.environ["KAFKA_SERVER"],
    sasl_plain_username=os.environ["KAFKA_BROKER_USERNAME"],
    sasl_plain_password=os.environ["KAFKA_BROKER_PASSWORD"],
    security_protocol="PLAINTEXT",
    sasl_mechanism="PLAINTEXT",
    value_serializer=lambda v: v.encode('utf-8')
)

其中KAFKA_SERVER的值为：gb-kafka.kafka.svc.gb.local:9092

错误堆栈信息 #

Traceback (most recent call last):
  File "/code/main.py", line 45, in <module>
    producer = KafkaProducer(
  File "/usr/local/lib/python3.10/site-packages/kafka/producer/kafka.py", line 381, in __init__
    client = KafkaClient(metrics=self._metrics, metric_group_prefix='producer',
  File "/usr/local/lib/python3.10/site-packages/kafka/client_async.py", line 244, in __init__
    self.config['api_version'] = self.check_version(timeout=check_timeout)
  File "/usr/local/lib/python3.10/site-packages/kafka/client_async.py", line 927, in check_version
    raise Errors.NoBrokersAvailable()
kafka.errors.NoBrokersAvailable: NoBrokersAvailable 

Kafka的Helm配置（values.yaml） #

使用bitnami/kafka的Helm Chart部署Kafka，仅修改了监听器相关配置：

listeners: "PLAINTEXT://:9092"
advertisedListeners: "PLAINTEXT://gb-kafka.kafka.svc.gb.local:9092"
listenerSecurityProtocolMap: "PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT"
allowPlaintextListener: true
interBrokerListenerName: PLAINTEXT

Kubernetes集群资源状态 #

执行kubectl get all的结果（IP已修改）：

NAME                       READY   STATUS    RESTARTS   AGE
pod/gb-kafka-0             1/1     Running   0          45m
pod/gb-kafka-zookeeper-0   1/1     Running   0          4d2h

NAME                                  TYPE        CLUSTER-IP      EXTERNAL-IP   PORT(S)                      AGE
service/gb-kafka                      ClusterIP   10.10.10.543    <none>        9092/TCP                     4d2h
service/gb-kafka-headless             ClusterIP   None            <none>        9092/TCP,9093/TCP            4d2h
service/gb-kafka-zookeeper            ClusterIP   10.222.01.220   <none>        2181/TCP,2888/TCP,3888/TCP   4d2h
service/gb-kafka-zookeeper-headless   ClusterIP   None            <none>        2181/TCP,2888/TCP,3888/TCP   4d2h

NAME                                  READY   AGE
statefulset.apps/gb-kafka             1/1     4d2h
statefulset.apps/gb-kafka-zookeeper   1/1     4d2h

解决方案 #

1. 调整FastAPI的Producer初始化时机与重试逻辑 #

FastAPI通常在启动时立即初始化Kafka Producer，此时Pod网络可能未完全就绪（DNS解析、网络策略生效存在延迟），而普通Python Pod的启动流程可能更慢，给了网络足够的初始化时间。

解决方法：

• 添加重试逻辑，用指数退避策略尝试连接：

import time
from kafka import KafkaProducer, errors

def create_kafka_producer():
    max_retries = 5
    retry_delay = 3
    for attempt in range(max_retries):
        try:
            return KafkaProducer(
                bootstrap_servers=os.environ["KAFKA_SERVER"],
                sasl_plain_username=os.environ["KAFKA_BROKER_USERNAME"],
                sasl_plain_password=os.environ["KAFKA_BROKER_PASSWORD"],
                security_protocol="PLAINTEXT",
                sasl_mechanism="PLAINTEXT",
                value_serializer=lambda v: v.encode('utf-8')
            )
        except errors.NoBrokersAvailable:
            if attempt < max_retries - 1:
                time.sleep(retry_delay)
                retry_delay *= 2
            else:
                raise

# 在FastAPI启动事件中初始化Producer
from fastapi import FastAPI

app = FastAPI()
producer = None

@app.on_event("startup")
async def startup_event():
    global producer
    producer = create_kafka_producer()

2. 检查FastAPI所在Namespace的网络策略 #

确认FastAPI所在Namespace是否有网络策略限制了对Kafka Namespace的9092端口访问，普通Python Pod可能所在Namespace无此限制。

解决方法：

• 查看当前网络策略：

kubectl get networkpolicies -n <fastapi-namespace>

• 如果存在限制，添加允许访问Kafka的网络策略：

apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: allow-kafka-access
  namespace: <fastapi-namespace>
spec:
  podSelector:
    matchLabels:
      app: <fastapi-app-label>
  policyTypes:
  - Egress
  egress:
  - to:
    - namespaceSelector:
        matchLabels:
          name: kafka
      podSelector:
        matchLabels:
          app.kubernetes.io/name: kafka
    ports:
    - protocol: TCP
      port: 9092

3. 验证环境变量注入是否正确 #

确认FastAPI Pod是否正确注入了KAFKA_SERVER、KAFKA_BROKER_USERNAME、KAFKA_BROKER_PASSWORD三个环境变量，部署时可能遗漏配置导致连接参数错误。

解决方法：

• 查看Pod环境变量：

kubectl exec -n <fastapi-namespace> <fastapi-pod-name> -- printenv | grep KAFKA_

• 若环境变量缺失，检查Deployment/StatefulSet配置，通过ConfigMap或Secret正确注入变量。

4. 延长Kafka客户端超时时间 #

FastAPI的Kafka客户端初始化超时时间可能过短，导致网络延迟时无法完成Broker版本检查。

解决方法：

• 初始化Producer时增加超时参数：

producer = KafkaProducer(
    bootstrap_servers=os.environ["KAFKA_SERVER"],
    sasl_plain_username=os.environ["KAFKA_BROKER_USERNAME"],
    sasl_plain_password=os.environ["KAFKA_BROKER_PASSWORD"],
    security_protocol="PLAINTEXT",
    sasl_mechanism="PLAINTEXT",
    value_serializer=lambda v: v.encode('utf-8'),
    api_version_auto_timeout_ms=30000  # 延长至30秒
)

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