Flink 使用介绍相关文档目录
背景
Flink Kerberos认证的配置位于YAML配置文件中，带来的便利是用户不用反复去配置认证信息。但这样使用有局限性。用户如果需要频繁切换不同的用户提交任务，是否可以像spark-submit那样通过--principal和--keytab参数来临时指定用户呢？答案是可以的。
使用方式
Flink 1.17版本支持使用-D 配置项的方式在提交任务时指定认证配置。例如：
./flink run-application -t yarn-application -D security.kerberos.login.keytab=/etc/security/keytabs/hdfs.headless.keytab -D security.kerberos.login.principal=hdfs@PAUL.COM ../examples/batch/WordCount.jar
其中-D 配置项的内容和YAML配置文件中的参数名相同。
需要注意的是，经本人验证Flink 1.13.2 和Flink 1.15.4版本存在问题，无法使用此方式。
经查询社区，和修复此问题相关联的Issue为：
- [FLINK-29435][client] SecurityConfiguration supports dynamic configuration
- [FLINK-31321][Deployment/YARN] Yarn-session mode, securityConfiguration supports dynamic configuration
如果需要在1.13.x和1.15.x版本中修复，可以将这两个commit cherrypick过去重新编译解决。
除此之外还有一个备用方法：动态指定Kerberos Cache。该方法对于Flink 1.13.x和1.15.x可用。
kinit hdfs@PAUL.COM -kt /etc/security/keytabs/hdfs.headless.keytab -c /opt/paul/krb_cache
export KRB5CCNAME=/opt/paul/krb_cache && ./flink run -m yarn-cluster ../examples/batch/WordCount.jar
kinit命令可以通过-c参数指定自定义的cache文件路径。在Flink使用的时候将其配置到KRB5CCNAME环境变量对应起来。不同用户使用不同的cache文件可避免相互影响。
KRB5CCNAME环境变量仅在当前shell中生效，不会干扰其他的shell提交作业。
源代码分析
security config动态参数解析部分
CliFrontend::mainInternal方法try块内容：
final CliFrontend cli = new CliFrontend(configuration, customCommandLines);
CommandLine commandLine =
cli.getCommandLine(
new Options(),
Arrays.copyOfRange(args, min(args.length, 1), args.length),
true);
Configuration securityConfig = new Configuration(cli.configuration);
// 获取命令行中-D key=value形式的动态参数，将其合并入securityConfig(Configuration类型)中
DynamicPropertiesUtil.encodeDynamicProperties(commandLine, securityConfig);
// 生成安全相关配置（Hadoop认证还是JAAS）
SecurityUtils.install(new SecurityConfiguration(securityConfig));
// 执行认证过程
retCode = SecurityUtils.getInstalledContext().runSecured(() -> cli.parseAndRun(args));
上面生成安全相关配置和执行认证过程参见Flink 源码之安全认证。
动态指定Kerberos Cache
相关源代码位于：KerberosUtils::static方法。相关代码片段为：
// 获取系统环境变量KRB5CCNAME
String ticketCache = System.getenv("KRB5CCNAME");
if (ticketCache != null) {
if (IBM_JAVA) {
System.setProperty("KRB5CCNAME", ticketCache);
} else {
// 添加到kerberosCacheOptions中
kerberosCacheOptions.put("ticketCache", ticketCache);
}
}
