首先用单机的java代码去消费kerberos认证的kafka，已经跑通，主要用到的几个文件如下所示：

1.jaas文件

 

2.krb5.conf

 

3.keytab认证文件

4.还需要提供principle，我的是kafka/bigdata-4@EXAMPLE.COM

单机的先把程序跑通了以后，就开始进行flink的认证。

首先需要修改flink配置文件flink-conf.yaml,修改这几项内容，然后把配置文件同步到整个集群，重启flink集群，内容如下

 

这是flink集群模块，另外还需要在代码里配置，代码里的配置不能只在主节点进行认证，因为实际消费kafka的时候是taskmanager消费的，所以需要在所有机器进行认证。代码如下：

主节点认证：

 

taskmanager认证：

 

消费kafka：

 

第二步的目的是为了把环境变量设置到其他机器上，经过测试，以上步骤，缺一不可。

需要注意的是，如果第二步的map操作并行度设置不当，那么会导致只有部分机器有kerberos认证，部分机器没有认证。

这一问题可以通过自定义kafka source来实现，在kafka source里面设置环境变量。