使用 Python 生产者和消费者在 Kubernetes minikube 上设置 Kafka Kraft

我正在尝试从 kubernetes 集群外部连接到 kubernetes minikube kafka pod。 服务器启动没有任何问题，但我无法设法将本地 kafka 生产者/消费者连接到外部 kafka pod。

在集群内的 kafka 服务器映像上，我将 bootstrap-server 设置为：

bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic "test"

服务kafka 服务器 pod 设置为：

apiVersion: v1
kind: Service
metadata:
  name: {{ .Chart.Name }}-service
spec:
  type: NodePort
  ports:
    - name: {{ .Chart.Name }}
      port: 9092
      nodePort: 30092
  selector:
    app: {{ .Chart.Name }}

我在部署中为侦听器设置的唯一 kafka 配置如下。

apiVersion: apps/v1
kind: Deployment
metadata:
  name: {{ .Chart.Name }}-deployment
  labels:
    app: {{ .Chart.Name }}
spec:
  {{- if not .Values.autoscaling.enabled }}
  replicas: {{ .Values.deployment.replicaCount }}
  {{- end }}
  selector:
    matchLabels:
      app: {{ .Chart.Name }}
  template:
    metadata:
      labels:
        app: {{ .Chart.Name }}
    spec:
      containers:
        - name: {{ .Chart.Name }}
          image: "{{ .Values.deployment.image }}"
          imagePullPolicy: {{ .Values.deployment.pullPolicy }}
          command: {{ .Values.deployment.initCommand | toYaml | nindent 12 }}
          ports:
              - containerPort: 9092
                name: internal
              - containerPort: 9093
                name: controller
              - containerPort: 30092
                name: nodeport
          env:
            - name: KRAFT_TOPIC_ML
              value: {{ .Values.deployment.topic.nameML }}
            - name: KRAFT_TOPIC_ML_REPLICATION_FACTOR
              value: "{{ .Values.deployment.topic.replicationFactorML }}"
            - name: KRAFT_TOPIC_ML_PARTITIONS
              value: "{{ .Values.deployment.topic.partitionsML }}"
            - name: KRAFT_TOPIC_MILP
              value: {{ .Values.deployment.topic.nameMILP }}
            - name: KRAFT_TOPIC_MILP_REPLICATION_FACTOR
              value: "{{ .Values.deployment.topic.replicationFactorMILP }}"
            - name: KRAFT_TOPIC_MILP_PARTITIONS
              value: "{{ .Values.deployment.topic.partitionsMILP }}"

            - name: KAFKA_LISTENER_SECURITY_PROTOCOL_MAP
              value: "INTERNAL:PLAINTEXT,NODEPORT:PLAINTEXT,CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL"
            - name: KAFKA_LISTENERS
              value: "INTERNAL://0.0.0.0:9092,NODEPORT://0.0.0.0:30092,CONTROLLER://0.0.0.0:9093"
            - name: KAFKA_ADVERTISED_LISTENERS
              value: "INTERNAL://localhost:9092,NODEPORT://localhost:30092"
            - name: KAFKA_INTER_BROKER_LISTENER_NAME
              value: "INTERNAL"
            - name: KAFKA_CONTROLLER_LISTENER
              value: "CONTROLLER"

当尝试使用本地 IP 和端口作为 30092 在我的计算机上本地连接到 kafka 服务器时可以找到代理并连接，但在生成消息时，它不会显示在相同 IP 和端口的消费者 python 端。

我尝试更改 kafka 侦听器配置，但没有成功。

你遇到的问题在于，尽管你将 KAFKA_ADVERTISED_LISTENERS 设置为使用 localhost ，但这个地址只能从集群内部访问。外部客户端（例如你的本地 Python 生产者和消费者）无法解析这个地址。

为了解决这个问题，你需要在 KAFKA_ADVERTISED_LISTENERS 中使用 Minikube 虚拟机的 IP 地址或外部可访问的域名。下面是如何操作的步骤：

1. 获取 Minikube IP 地址:

在你的终端中运行以下命令：

bash minikube ip

这将返回 Minikube 虚拟机的 IP 地址（例如，192.168.99.100）。

2. 更新 Kafka 配置:

将 KAFKA_ADVERTISED_LISTENERS 的值更新为使用 Minikube IP 地址：

yaml KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka-service:9092,NODEPORT://<MINIKUBE_IP>:30092

将 <MINIKUBE_IP> 替换为你在步骤 1 中获取的 IP 地址。

注意:

• 我们将内部侦听器配置为使用服务名称 kafka-service 进行 pod 间通信。
• 确保 NODEPORT 侦听器使用 Minikube IP 和正确的 NodePort（在本例中为 30092）。

3. 应用更改:

将更新后的部署配置应用于你的 Kubernetes 集群：

bash kubectl apply -f <your-deployment-file.yaml>

4. 使用 Minikube IP 连接:

现在，在你的 Python 生产者和消费者代码中，使用 Minikube IP 地址和 NodePort 连接到 Kafka：

python bootstrap_servers=['<MINIKUBE_IP>:30092']

完成这些更改后，你的本地 Python 生产者和消费者应该能够连接到在 Minikube 上运行的 Kafka 集群并与之交互。