如何在Kubernetes的Kafka部署启动时引入kafka-avro-console-producer？

1. 先部署Schema Registry到Kubernetes #

Schema Registry需与Kafka集群通信，先准备Deployment和Service配置：

• Deployment配置：使用Confluent官方镜像confluentinc/cp-schema-registry，环境变量指定Kafka bootstrap地址、Schema Registry监听地址及主机名：

  apiVersion: apps/v1
  kind: Deployment
  metadata:
    name: schema-registry
  spec:
    replicas: 1
    selector:
      matchLabels:
        app: schema-registry
    template:
      metadata:
        labels:
          app: schema-registry
      spec:
        containers:
        - name: schema-registry
          image: confluentinc/cp-schema-registry:7.5.0
          ports:
          - containerPort: 8081
          env:
          - name: SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS
            value: "your-kafka-service:9092" # 替换为你的Kafka Service地址
          - name: SCHEMA_REGISTRY_LISTENERS
            value: "http://0.0.0.0:8081"
          - name: SCHEMA_REGISTRY_HOST_NAME
            value: schema-registry-service
  

• Service配置：创建ClusterIP类型的Service，让K8s内部组件可访问Schema Registry：

  apiVersion: v1
  kind: Service
  metadata:
    name: schema-registry-service
  spec:
    selector:
      app: schema-registry
    ports:
    - port: 8081
      targetPort: 8081
  

2. 解决kafka-avro-console-producer脚本的可用性问题 #

你之前看到脚本在kafka/bin下，是因为当时用的是Confluent Platform的Kafka镜像（confluentinc/cp-kafka），而非Apache Kafka官方镜像。有两种可行方案：

方案一：替换现有Kafka镜像为Confluent版 #

修改你的Kafka Deployment，将镜像从apache/kafka替换为confluentinc/cp-kafka（需指定与Schema Registry兼容的版本）。该镜像自带kafka-avro-console-producer，路径通常为/usr/bin/kafka-avro-console-producer或/opt/kafka/bin/。

• 进入Kafka Pod执行命令：

  kubectl exec -it <kafka-pod-name> -- /bin/bash
  kafka-avro-console-producer --broker-list your-kafka-service:9092 --topic test-topic --property schema.registry.url=http://schema-registry-service:8081
  

方案二：单独启动包含Avro工具的Pod #

若不想替换现有Kafka镜像，可单独启动一个承载Confluent工具的Pod：

apiVersion: v1
kind: Pod
metadata:
  name: kafka-avro-tools
spec:
  containers:
  - name: kafka-avro-tools
    image: confluentinc/cp-kafka:7.5.0
    command: ["sleep", "3600"] # 保持Pod持续运行
    env:
    - name: SCHEMA_REGISTRY_URL
      value: "http://schema-registry-service:8081"

创建后进入Pod使用脚本：

kubectl exec -it kafka-avro-tools -- /bin/bash
# 示例：发送Avro格式消息
kafka-avro-console-producer --broker-list your-kafka-service:9092 --topic test-topic \
  --property schema.registry.url=http://schema-registry-service:8081 \
  --property value.schema='{"type":"record","name":"Test","fields":[{"name":"id","type":"int"},{"name":"name","type":"string"}]}'

3. 验证连通性 #

确保Schema Registry与Kafka正常通信：
从Kafka Pod或工具Pod执行：

curl http://schema-registry-service:8081/subjects

若返回空数组或已存在的subject列表，则说明连接正常。

注意事项 #

• 版本兼容：Confluent各组件（Kafka、Schema Registry）版本需保持一致，避免兼容性问题。
• 网络访问：确保Kafka Pod、工具Pod能访问Schema Registry Service，Kafka集群地址配置正确。

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