ambari安装的kafka，启用了kerberos，在使用kafka-console-producer时报错

ambari安装kafka，共3台broker（hdp03-05），3台zk（hdp01-03），hdp版本是3.1.5，kafka是2.0.0，使用kafka-console-producer.sh时会报错，但使用kafka-topics.sh正常。

主要配置如下：
server.properties

advertised.listeners=SASL_PLAINTEXT://hdp03.hdp.com:6667
listeners=SASL_PLAINTEXT://hdp03.hdp.com:6667
sasl.enabled.mechanisms=GSSAPI
sasl.mechanism.inter.broker.protocol=GSSAPI
security.inter.broker.protocol=SASL_PLAINTEXT
zookeeper.connect=hdp01.hdp.com:2181,hdp02.hdp.com:2181,hdp03.hdp.com:2181

kafka_jaas.conf

KafkaServer {
    com.sun.security.auth.module.Krb5LoginModule required
    useKeyTab=true
    keyTab="/etc/security/keytabs/kafka.service.keytab"
    storeKey=true
    useTicketCache=false
    serviceName="kafka"
    principal="kafka/hdp03.hdp.com@HDP315.COM";
};
KafkaClient {
    com.sun.security.auth.module.Krb5LoginModule required
    useTicketCache=true
    renewTicket=true
    serviceName="kafka";
};
Client {
    com.sun.security.auth.module.Krb5LoginModule required
    useKeyTab=true
    keyTab="/etc/security/keytabs/kafka.service.keytab"
    storeKey=true
    useTicketCache=false
    serviceName="zookeeper"
    principal="kafka/hdp03.hdp.com@HDP315.COM";
};
com.sun.security.jgss.krb5.initiate {
    com.sun.security.auth.module.Krb5LoginModule required
    renewTGT=false
    doNotPrompt=true
    useKeyTab=true
    keyTab="/etc/security/keytabs/kafka.service.keytab"
    storeKey=true
    useTicketCache=false
    serviceName="kafka"
    principal="kafka/hdp03.hdp.com@HDP315.COM";
};

kafka_client_jaas.conf内容：

KafkaClient {
    com.sun.security.auth.module.Krb5LoginModule required
    useTicketCache=true
    renewTicket=true
    serviceName="kafka";
};

producer.properties内容：

security.protocol=SASL_PLAINTEXT
sasl.mechanism=GSSAPI
sasl.kerberos.service.name=kafka

kafka-console-producer.sh内容：

if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
    export KAFKA_HEAP_OPTS="-Xmx512M"
fi

# check if kafka_jaas.conf in config , only enable client_kerberos_params in secure mode.
KAFKA_HOME="$(dirname $(cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd ))"
KAFKA_JAAS_CONF=$KAFKA_HOME/config/kafka_jaas.conf
if [ -f $KAFKA_JAAS_CONF ]; then
    export KAFKA_CLIENT_KERBEROS_PARAMS="-Djava.security.auth.login.config=$KAFKA_HOME/config/kafka_client_jaas.conf"
fi

exec $(dirname $0)/kafka-run-class.sh kafka.tools.ConsoleProducer "$@"

在使用kafka-console-producer.sh时报错认证失败

kinit -kt /etc/security/keytabs/kafka.service.keytab kafka/hdp03.hdp.com@HDP315.COM
./kafka-console-producer.sh --broker-list hdp03.hdp.com:6667,hdp04.hdp.com:6667,hdp05.hdp.com:6667 --topic test-topic --producer.config /root/producer.properties

报错信息如下：

[2023-04-10 09:05:09,740] ERROR [Producer clientId=console-producer] Connection to node -2 failed authentication due to: Authentication failed due to invalid credentials with SASL mechanism GSSAPI (org.apache.kafka.clients.NetworkClient)
[2023-04-10 09:05:09,740] ERROR Error when sending message to topic test-topic with key: null, value: 0 bytes with error: (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)
org.apache.kafka.common.errors.SaslAuthenticationException: Authentication failed due to invalid credentials with SASL mechanism GSSAPI
[2023-04-10 09:05:09,741] ERROR [Producer clientId=console-producer] Connection to node -3 failed authentication due to: Authentication failed due to invalid credentials with SASL mechanism GSSAPI (org.apache.kafka.clients.NetworkClient)
>[2023-04-10 09:05:09,748] ERROR [Producer clientId=console-producer] Connection to node -1 failed authentication due to: Authentication failed due to invalid credentials with SASL mechanism GSSAPI (org.apache.kafka.clients.NetworkClient)
[2023-04-10 09:05:09,858] ERROR [Producer clientId=console-producer] Connection to node -3 failed authentication due to: Authentication failed due to invalid credentials with SASL mechanism GSSAPI (org.apache.kafka.clients.NetworkClient)
[2023-04-10 09:05:09,858] ERROR [Producer clientId=console-producer] Connection to node -1 failed authentication due to: Authentication failed due to invalid credentials with SASL mechanism GSSAPI (org.apache.kafka.clients.NetworkClient)

还请各位帮忙指点问题所在

不知道你这个问题是否已经解决, 如果还没有解决的话:

• 这篇文章：Kafka Producer 发送大消息该如何配置以及如何提高吞吐量 也许有你想要的答案，你可以看看
• 除此之外, 这篇博客: kafka使用kafka-console-consumer.sh和kafka-console-producer.sh生产消费数据样例中的 kafka-console-producer.sh生产有key消息 部分也许能够解决你的问题, 你可以仔细阅读以下内容或者直接跳转源博客中阅读:
  

  进入到kafka的bin目录下：执行以下脚本：

  ./kafka-console-producer.sh --broker-list List<host:port> --topic topicName --property parse.key=true

  默认消息key与消息value间使用“Tab键”进行分隔，所以消息key以及value中切勿使用转义字符(\t)

  kafkaKey    kafkaValue

  如上图所示，kafkaKey为消息key，kafkaValue为消息value，输入之后点击enter键即完成数据的生产

如果你已经解决了该问题, 非常希望你能够分享一下解决方案, 写成博客, 将相关链接放在评论区, 以帮助更多的人 ^-^