抱歉，漏了文档链接
https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/connectors/kafka.html#enabling-kerberos-authentication
Shuiqiang Chen <acqua....@gmail.com> 于2021年1月30日周六 下午4:32写道：
> Hi,
> 按照这个文档， 在flink-conf.yaml里配置了security.kerberos.login.keytab 和
> security.kerberos.login.principal这两个属性了吗？
> 还有就是jaas.conf文件在各个task manager所在的机器都能访问到吗？
>
> 瞿叶奇 <389243...@qq.com> 于2021年1月30日周六 下午4:15写道：
>
>> 老师，您好，
>> 1）报错信息在附件内，flink-root-python-node-master1aSdM.log文件。报错信息如下：
>> Caused by: java.lang.IllegalArgumentException: Could not find a
>> 'KafkaClient' entry in the JAAS configuration. System property
>> 'java.security.auth.login.config' is not set at
>> org.apache.kafka.common.security.JaasContext.defaultContext(JaasContext.java:133)
>> 2）这是Flink配置里的jaas.conf
>> 3)pyspark对接kafka写hdfs没有问题，这是spark的jaas.conf
>> 3.1）这是spark配置里的jaas.conf
>>
>> 3.2）这是spark配置里的jaas-zk.conf
>>
>>
>> 是不是要修改这个文件呢？百度给出的都是java的修正，不知道这个需要如何修改。
>>
>>
>> ------------------ 原始邮件 ------------------
>> *发件人:* "user-zh" <acqua....@gmail.com>;
>> *发送时间:* 2021年1月30日(星期六) 下午3:49
>> *收件人:* "user-zh"<user-zh@flink.apache.org>;
>> *主题:* Re: 问题求助（Pyflink）
>>
>> 你好，
>> 可以看下source task所在task manager 的日志里看看 consumer 有没有成功获取到kafka
>> partition相关meta信息和认证相关是否成功的信息。
>>
>> 瞿叶奇 <389243...@qq.com> 于2021年1月30日周六 下午3:14写道：
>>
>> > 老师，你好，消费是没有任何问题，可以正常消费。
>> >
>> >
>> >
>> >
>> > ------------------ 原始邮件 ------------------
>> > *发件人:* "user-zh" <appleyu...@163.com>;
>> > *发送时间:* 2021年1月30日(星期六) 下午3:08
>> > *收件人:* "user-zh"<user-zh@flink.apache.org>;
>> > *主题:* Re:问题求助（Pyflink）
>> >
>> > 先看下kafka能否通过命令行消费数据.
>> >
>> > 命令行检查确保能消费,再使用Flink.
>> >
>> >
>> >
>> >
>> >
>> >
>> >
>> >
>> >
>> >
>> >
>> >
>> >
>> > 在 2021-01-30 14:25:57，"瞿叶奇" <389243...@qq.com> 写道：
>> >
>> >
>> 老师，您好，我想通过Flink消费kafka写本地csv文件，目前遇到的问题是Flink、Kafka都是kerberos认证的集群，而且我是用的是pyflink，现在程序能执行，不报错，但是不消费数据，csv文件没有结果，但是文件日期修改日期一直在更新。怀疑对接kafka
>> > 存在问题，希望老师能够给解决疑惑。
>> > 1）Kafka生产数据：
>> >
>> > 2）pyflink 程序
>> >
>> >
>> > #!/usr/bin/python3.7
>> > # -*- coding: UTF-8 -*-
>> > from pyflink.datastream import StreamExecutionEnvironment,
>> > CheckpointingMode
>> > from pyflink.table import StreamTableEnvironment, TableConfig,
>> DataTypes,
>> > CsvTableSink, WriteMode, SqlDialect
>> > from pyflink.table.descriptors import
>> FileSystem,OldCsv,Schema,Kafka,Json
>> > s_env = StreamExecutionEnvironment.get_execution_environment()
>> > s_env.set_parallelism(1)
>> > s_env.enable_checkpointing(3000)
>> > st_env = StreamTableEnvironment.create(s_env, TableConfig())
>> > st_env.use_catalog("default_catalog")
>> > st_env.use_database("default_database")
>> >
>> st_env.connect(Kafka().version("universal").topic("qyq13").start_from_earliest().property("zookeeper.connect",
>> > "192.168.0.120:24002,192.168.0.238:24002,192.168.0.6:24002
>> ").property("security.protocol",
>> > 'SASL_PLAINTEXT').property("sasl.kerberos.service.name",
>> > 'kafka').property("kerberos.domain.name", 'hadoop.hadoop.com
>> ').property("bootstrap.servers",
>> > "192.168.0.151:21007,192.168.0.29:21007,192.168.0.93:21007
>> ")).with_format(Json().fail_on_missing_field(False).schema(DataTypes.ROW([DataTypes.FIELD("id",
>> > DataTypes.BIGINT()),DataTypes.FIELD("name",
>> > DataTypes.STRING())]))).with_schema(Schema().field("id",
>> > DataTypes.BIGINT()).field("name",
>> > DataTypes.STRING())).register_table_source("sourceKafka")
>> > fieldNames = ["id", "name"]
>> > fieldTypes = [DataTypes.BIGINT(),DataTypes.STRING()]
>> > csvSink = CsvTableSink(fieldNames, fieldTypes, "/tmp/result2021.csv",
>> ",",
>> > 1, WriteMode.OVERWRITE)
>> > st_env.register_table_sink("csvTableSink", csvSink)
>> > resultQuery = st_env.sql_query("select id,name from sourceKafka")
>> > resultQuery = resultQuery.insert_into("csvTableSink")
>> > st_env.execute("pyflink-kafka-v2")
>> > 3）pyflink-shell.sh local
>> >
>> > 4)运行结果
>> > 在pyflink-shell local运行程序的同时，启用生产者生产数据，查看结果文件如下：
>> >
>> >
>> >
>> 可以看出文件确实在被更新，文件的修改时间在变化，但是里面是空的，一方面希望pyflink可以增加打印到控制台的功能，一方面希望老师能给出对接kerberos认证的kafka的案例，我是陕西国网用电信息采集系统新架构改造的工程师，我们计划使用flink测试kafka-hdfs数据的数据传输。希望老师能给出一个案例，帮助我们完成测试。
>> >
>> >
>> >
>> >
>>
>
