我想建议你做一些尝试。
设置Kafka选项
export kafka\u opts=“-djava.security.auth.login.config=/etc/kafka/kafka\u sink\u jaas.conf”
在代码中添加以下内容

System.setProperty("java.security.auth.login.config", "/etc/kafka/kafka_sink_jaas.conf")
System.setProperty("java.security.krb5.conf", "/etc/krb5.conf")
System.setProperty("sun.security.krb5.debug", false)
System.setProperty("javax.security.auth.useSubjectCredsOnly", "false")

希望这会有帮助。

在本地运行flink时，我使用了这个jaas配置文件。如fred所述，以下命令在集群模式下不起作用。

KafkaClient {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="MyUserId"
    password="MyPwd";
    };

security.kerberos.login.use-ticket-cache: false 
security.kerberos.login.keytab: /etc/kafka/kafka.keytab
security.kerberos.login.principal: kafka@HADOOP.COM
security.kerberos.login.contexts: Client,KafkaClient

如果我们使用kerberos设置，这是正确的吗？security.kerberos.login.principal:myuserid
为了创建一个keytab文件，我使用了klist。它似乎总是添加一个域名与用户例如。myuserid@somedomain.com. 有没有办法避免这种情况？

我已经解决了这个问题。它是由flink集群kerberos配置引起的。只需要在flink-conf上设置一些配置，yaml就可以让它正常工作！
设置如下：

security.kerberos.login.use-ticket-cache: false 
security.kerberos.login.keytab: /etc/kafka/kafka.keytab
security.kerberos.login.principal: kafka@HADOOP.COM
security.kerberos.login.contexts: Client,KafkaClient