背景

现在假设k8s中已经启动了kafka容器
kafka所在Namespace为xxx

1、启动一个kafka client的pod

kubectl run kafka-client --restart=‘Never’ --image bitnami/kafka:3.1.0-debian-10-r85 --namespace xxx --command – sleep infinity

2、配置

运行脚本生成client.properties

# 1.从 secret 中获取kafka 密码
password="$(kubectl get secret kafka-jaas --namespace xxx -o jsonpath='{.data.client-passwords}' | base64 --decode | cut -d , -f 1)"
# 2.创建配置文件： client.properties
echo "security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username='qtUser' password='$password';" > client.properties

将配置文件copy到client容器

kubectl cp --namespace xxx client.properties kafka-client:/tmp/client.properties

3、进入client

kubectl exec --tty -i kafka-client --namespace xxx -- bash

4、测试

kafka的client可以直接发送消息和接受消息
发送消息

kafka-console-producer.sh \
            --producer.config /tmp/client.properties \
            --broker-list kafka-0.kafka-headless.xxx.svc.cluster.local:9092 \
            --topic test

接受消息

• 可以用来测试项目中已有的消息，因为kafka的消息是可以多次消费的，这里指定了从头开始。

kafka-console-consumer.sh \
            --consumer.config /tmp/client.properties \
            --bootstrap-server kafka-headless.xxx:9092 \
            --topic test \
            --from-beginning

列出所有topic

kafka-topics.sh --list --bootstrap-server kafka-headless.xxx:9092  --command -config /tmp/client.properties