【Flink】Flink跨集群访问开启Kerberos认证的Kafka

转x33g5p2x  于2022-08-17 转载在 Flink  
字(17.8k)|赞(0)|评价(0)|浏览(1615)

1.概述

转载：Flink跨集群访问开启Kerberos认证的Kafka

Flink提供了三个模块来对集群进行安全验证，分别是HadoopModule、JaasModule、ZooKeeperModule。安全认证相关参数对应的类SecurityOptions。

HadoopModule用来对使用UserGroupInformation进行身份验证的框架（kudu、hbase同步框架、hdfs等）进行认证配置。
JaasModule用来对使用JaasConfig进行身份验证的框架（kafka、zk、hbase异步框架等）进行认证配置。
ZooKeeperModule负责安装整个进程的ZooKeeper安全配置。

Flink组件在启动时，会先加载认证相关模块，在构建的安全上下文中，启动集群各个组件。不过Flink整个集群只能使用一份证书进行相关验证，也就是说，如果Flink任务从开启Kerberos认证的Kafka中读取数据，并写入Kudu，则使用的principal和keytab，具有同时访问hdfs、kafka、kudu的权限。如果使用不同的证书，则需要在Flink任务中单独进行Kerberos相关配置。

2.启动流程

以JM启动为例，查看加载安全认证相关组件，记录各个模块调用链。

加载安全认证上下文，从上下文中启动集群各个组件。

  1. ClusterEntrypoint#startCluster
  2. // 
  3. SecurityContext securityContext = installSecurityContext(configuration);
  4. securityContext.runSecured((Callable<Void>) () -> {
  5.     runCluster(configuration);
  6.     return null;
  7. });
  9. SecurityContext installSecurityContext(Configuration configuration)） {
  10.     SecurityUtils.install(new SecurityConfiguration(configuration));
  11.     return SecurityUtils.getInstalledContext();
  12. }

SecurityConfiguration默认提供了两个security.context.factory.classes用来构建SecurityContext：
org.apache.flink.runtime.security.contexts.HadoopSecurityContextFactory：根据构建的UserGroupInformation在doAs方法中启动集群。
org.apache.flink.runtime.security.contexts.NoOpSecurityContextFactory：默认在不需要进行安全认证的上下文中启动。
提供三个security.module.factory.classes用来准备安全认证使用的配置：

  1. org.apache.flink.runtime.security.modules.HadoopModuleFactory
  2. org.apache.flink.runtime.security.modules.JaasModuleFactory
  3. org.apache.flink.runtime.security.modules.ZookeeperModuleFactory

主要还是installModules配置认证使用的配置

  1. SecurityUtils#install
  2. public static void install(SecurityConfiguration config) throws Exception {
  3.    // Install the security modules first before installing the security context
  4.    installModules(config);
  5.    installContext(config);
  6. }
  8. #installModules使用SPI动态创建moduleFactory，分别调用其install方法
  9. static void installModules(SecurityConfiguration config) {
  10.     List<SecurityModule> modules = new ArrayList<>();
  11.     for (String moduleFactoryClass : config.getSecurityModuleFactories()) {
  12.         SecurityModuleFactory moduleFactory = null;
  13.         SecurityModule module = moduleFactory.createModule(config);
  14.             if (module != null) {
  15.                 module.install();
  16.                 modules.add(module);
  17.             }
  18.         }
  19.         installedModules = modules;
  20.     }
  21. }

HadoopModule#install。hdfs进行kerberos认证需要UserGroupInformation作为loginUser，该模块用来构建全局的loginUser，如果其他组件能够使用该loginUser进行认证，则不需要单独配置证书。

  1. public void install() throws SecurityInstallException {
  2.     ## 传递hadoop相关参数
  3.         UserGroupInformation.setConfiguration(hadoopConfiguration);
  5.     UserGroupInformation loginUser;
  7.     try {
  8.         ## 开启kerberos认证并传递Keytab和Principal
  9.             if (UserGroupInformation.isSecurityEnabled() &&
  10.                 !StringUtils.isBlank(securityConfig.getKeytab()) && !StringUtils.isBlank(securityConfig.getPrincipal())) {
  11.                 String keytabPath = (new File(securityConfig.getKeytab())).getAbsolutePath();
  13.                 UserGroupInformation.loginUserFromKeytab(securityConfig.getPrincipal(), keytabPath);
  14.                 ## 当前登陆用户
  15.                     loginUser = UserGroupInformation.getLoginUser();
  16.                 //  token cache
  17.                 // supplement with any available tokens
  18.                 String fileLocation = System.getenv(UserGroupInformation.HADOOP_TOKEN_FILE_LOCATION);
  19.                 if (fileLocation != null) {
  20.                     // Use reflection API since the API semantics are not available in Hadoop1 profile. Below APIs are
  21.                     // used in the context of reading the stored tokens from UGI.
  22.                     // Credentials cred = Credentials.readTokenStorageFile(new File(fileLocation), config.hadoopConf);
  23.                     // loginUser.addCredentials(cred);
  24.                     try {
  25.                         Method readTokenStorageFileMethod = Credentials.class.getMethod("readTokenStorageFile",
  26.                                                                                         File.class, org.apache.hadoop.conf.Configuration.class);
  27.                         Credentials cred =
  28.                             (Credentials) readTokenStorageFileMethod.invoke(
  29.                             null,
  30.                             new File(fileLocation),
  31.                             hadoopConfiguration);
  33.                         // if UGI uses Kerberos keytabs for login, do not load HDFS delegation token since
  34.                         // the UGI would prefer the delegation token instead, which eventually expires
  35.                         // and does not fallback to using Kerberos tickets
  36.                         Method getAllTokensMethod = Credentials.class.getMethod("getAllTokens");
  37.                         Credentials credentials = new Credentials();
  38.                         final Text hdfsDelegationTokenKind = new Text("HDFS_DELEGATION_TOKEN");
  39.                         Collection<Token<? extends TokenIdentifier>> usrTok = (Collection<Token<? extends TokenIdentifier>>) getAllTokensMethod.invoke(cred);
  40.                         //If UGI use keytab for login, do not load HDFS delegation token.
  41.                         for (Token<? extends TokenIdentifier> token : usrTok) {
  42.                             if (!token.getKind().equals(hdfsDelegationTokenKind)) {
  43.                                 final Text id = new Text(token.getIdentifier());
  44.                                 credentials.addToken(id, token);
  45.                             }
  46.                         }
  48.                         Method addCredentialsMethod = UserGroupInformation.class.getMethod("addCredentials",
  49.                                                                                            Credentials.class);
  50.                         addCredentialsMethod.invoke(loginUser, credentials);
  51.                     } catch (NoSuchMethodException e) {
  52.                         LOG.warn("Could not find method implementations in the shaded jar.", e);
  53.                     } catch (InvocationTargetException e) {
  54.                         throw e.getTargetException();
  55.                     }
  56.                 }
  57.             } else {
  58.                 // login with current user credentials (e.g. ticket cache, OS login)
  59.                 // note that the stored tokens are read automatically
  60.                 try {
  61.                     //Use reflection API to get the login user object
  62.                     //UserGroupInformation.loginUserFromSubject(null);
  63.                     Method loginUserFromSubjectMethod = UserGroupInformation.class.getMethod("loginUserFromSubject", Subject.class);
  64.                     loginUserFromSubjectMethod.invoke(null, (Subject) null);
  65.                 } catch (NoSuchMethodException e) {
  66.                     LOG.warn("Could not find method implementations in the shaded jar.", e);
  67.                 } catch (InvocationTargetException e) {
  68.                     throw e.getTargetException();
  69.                 }
  71.                 loginUser = UserGroupInformation.getLoginUser();
  72.             }
  74.         boolean isCredentialsConfigured = HadoopUtils.isCredentialsConfigured(
  75.             loginUser, securityConfig.useTicketCache());
  77.         LOG.info("Hadoop user set to {}, credentials check status: {}", loginUser, isCredentialsConfigured);
  79.     } catch (Throwable ex) {
  80.         throw new SecurityInstallException("Unable to set the Hadoop login user", ex);
  81.     }
  82. }

JaasModule#install。准备jaas文件使用的各个属性，先传递给ConfigFile，当各个框架使用jaas文件进行验证时，从javax.security.auth.login.Configuration中提取。

Kafka kerberos认证使用的jaas文件：

  1. KafkaClient {
  2.         com.sun.security.auth.module.Krb5LoginModule required
  3.         keyTab="/Users/xx/kafka.keytab"
  4.         principal="kafka/cdh002@TEST.COM"
  5.         useKeyTab=true
  6.         useTicketCache=true;
  7.  };

jaas文件使用的参数，写入javax.security.auth.login.Configuration。各个框架使用时会从改配置中取。

  1. public void install() {
  2.     priorConfigFile = System.getProperty(JAVA_SECURITY_AUTH_LOGIN_CONFIG, null);
  3.     ## 创建一个空的jaas文件，和环境变量java.security.auth.login.config绑定
  4.         if (priorConfigFile == null) {
  5.             File configFile = generateDefaultConfigFile(workingDir);
  6.             System.setProperty(JAVA_SECURITY_AUTH_LOGIN_CONFIG, configFile.getAbsolutePath());
  7.             LOG.info("Jaas file will be created as {}.", configFile);
  8.         }
  9.     // read the JAAS configuration file,    创建ConfigFile
  10.     priorConfig = javax.security.auth.login.Configuration.getConfiguration();
  12.     // construct a dynamic JAAS configuration
  13.     currentConfig = new DynamicConfiguration(priorConfig);
  15.     // wire up the configured JAAS login contexts to use the krb5 entries
  16.     AppConfigurationEntry[] krb5Entries = getAppConfigurationEntries(securityConfig);
  17.     if (krb5Entries != null) {
  18.         for (String app : securityConfig.getLoginContextNames()) {
  19.             currentConfig.addAppConfigurationEntry(app, krb5Entries);   // kafkaClient
  20.         }
  21.     }
  22.     ## 写入javax.security.auth.login.Configuration
  23.         javax.security.auth.login.Configuration.setConfiguration(currentConfig);
  24. }

ZooKeeperModule#install

  1. public void install() throws SecurityInstallException {
  3.     priorSaslEnable = System.getProperty(ZK_ENABLE_CLIENT_SASL, null);
  4.     System.setProperty(ZK_ENABLE_CLIENT_SASL, String.valueOf(!securityConfig.isZkSaslDisable()));
  6.     priorServiceName = System.getProperty(ZK_SASL_CLIENT_USERNAME, null);
  7.     if (!"zookeeper".equals(securityConfig.getZooKeeperServiceName())) {
  8.         System.setProperty(ZK_SASL_CLIENT_USERNAME, securityConfig.getZooKeeperServiceName());
  9.     }
  11.     priorLoginContextName = System.getProperty(ZK_LOGIN_CONTEXT_NAME, null);
  12.     if (!"Client".equals(securityConfig.getZooKeeperLoginContextName())) {
  13.         System.setProperty(ZK_LOGIN_CONTEXT_NAME, securityConfig.getZooKeeperLoginContextName());
  14.     }
  15. }

加载完各个Module后，构建installContext。只会有一个installedContext。

  1. static void installContext(SecurityConfiguration config) throws Exception {
  2.         // install the security context factory
  3.         for (String contextFactoryClass : config.getSecurityContextFactories()) {
  4.             try {
  5.                 // spi加载
  6.                 SecurityContextFactory contextFactory = SecurityFactoryServiceLoader.findContextFactory(contextFactoryClass);
  7.                 //  有hadoop环境就走Hadoop 认证
  8.                 if (contextFactory.isCompatibleWith(config)) {   
  9.                     try {
  10.                         installedContext = contextFactory.createContext(config);
  11.                         // install the first context that's compatible and ignore the remaining.   只加载一个
  12.                         break;
  13.                     } catch (SecurityContextInitializeException e) {
  14.                         LOG.error("Cannot instantiate security context with: " + contextFactoryClass, e);
  15.                     }
  16.                 } else {
  17.                     LOG.warn("Unable to install incompatible security context factory {}", contextFactoryClass);
  18.                 }
  19.             } catch (NoMatchSecurityFactoryException ne) {
  20.                 LOG.warn("Unable to instantiate security context factory {}", contextFactoryClass);
  21.             }
  22.         }
  23.         if (installedContext == null) {
  24.             LOG.error("Unable to install a valid security context factory!");
  25.             throw new Exception("Unable to install a valid security context factory!");
  26.         }
  27.     }

Flink读取开启Kerberos认证的kafka时，需要进行如下配置。并未传递java.security.auth.login.config以及sasl.jaas.config配置。

  1. kafka中添加的参数：
  2.         1. security.protocol='SASL_PLAINTEXT'  //使用SASL认证协议
  3.         2. sasl.mechanism = 'GSSAPI'           // 使用kerberos认证
  4.         3. sasl.kerberos.service.name = 'kafka' // 服务名称
  5. flinkconf.yaml中参数：
  6.         1. security.kerberos.login.use-ticket-cache:true
  7.         2. security.kerberos.login.keytab: xxxx
  8.         3. security.kerberos.login.principal:xxxxx
  9.         4. security.kerberos.login.contexts: KafkaClient,Client

如果配置sasl.jaas.config，则格式为：

  1. String config = "com.sun.security.auth.module.Krb5LoginModule  required\n" +
  2.     "\tprincipal=\"kafka/xxxxx@EXAMPLE.COM\"\n" +
  3.     "\tkeyTab=\"/Users/xxxx/kafka.keytab\"\n" +
  4.     "\tuseKeyTab=true\n" +
  5.     "\tuseTicketCache=true;";

Flink kafka connector 使用的jaas配置流程。

  1. FlinkKafkaConsumerBase#open
  2.     |
  3.     this.partitionDiscoverer.open();
  4.     |
  5.     KafkaPartitionDiscoverer#initializeConnections
  6.     |
  7. KafkaConsumer#KafkaConsumer
  8.     |
  9.     ClientUtils.createChannelBuilder(config);
  10.     |
  11.     ChannelBuilders#clientChannelBuilder
  12.     |
  13.     ChannelBuilders#create
  14.     |
  15.     SaslChannelBuilder（0.10版本）#configure
  16.     |
  17.     JaasUtils#jaasConfig
  1. public static Configuration jaasConfig(LoginType loginType, Map<String, ?> configs) {
  2.     Password jaasConfigArgs = (Password) configs.get(SaslConfigs.SASL_JAAS_CONFIG);
  3.     # 绑定了sasl.jaas.config参数，则从val中获取
  4.     if (jaasConfigArgs != null) {
  5.         if (loginType == LoginType.SERVER)
  6.             throw new IllegalArgumentException("JAAS config property not supported for server");
  7.         else {
  8.             JaasConfig jaasConfig = new JaasConfig(loginType, jaasConfigArgs.value());
  9.             AppConfigurationEntry[] clientModules = jaasConfig.getAppConfigurationEntry(LoginType.CLIENT.contextName());
  10.             int numModules = clientModules == null ? 0 : clientModules.length;
  11.             if (numModules != 1)
  12.                 throw new IllegalArgumentException("JAAS config property contains " + numModules + " login modules, should be one module");
  13.             return jaasConfig;
  14.         }
  15.     } else
  16.      
  17.         return defaultJaasConfig(loginType);
  18. }
  20. private static Configuration defaultJaasConfig(LoginType loginType) {
  21.     # 从Flink jaasModule加载时，已经绑定java.security.auth.login.config
  22.     String jaasConfigFile = System.getProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM);
  23.     if (jaasConfigFile == null) {
  24.         LOG.debug("System property '" + JaasUtils.JAVA_LOGIN_CONFIG_PARAM + "' and Kafka SASL property '" +
  25.                   SaslConfigs.SASL_JAAS_CONFIG + "' are not set, using default JAAS configuration.");
  26.     }
  27.     #  拿到javax.security.auth.login.Configuration中的配置,jaas中使用的认证实体
  28.     Configuration jaasConfig = Configuration.getConfiguration();
  29.     # KafkaClient
  30.     String loginContextName = loginType.contextName();
  31.     AppConfigurationEntry[] configEntries = jaasConfig.getAppConfigurationEntry(loginContextName);
  32.     if (configEntries == null) {
  33.         String errorMessage;
  34.         errorMessage = "Could not find a '" + loginContextName + "' entry in the JAAS configuration. System property '" +
  35.             JaasUtils.JAVA_LOGIN_CONFIG_PARAM + "' is " + (jaasConfigFile == null ? "not set" : jaasConfigFile);
  36.         throw new IllegalArgumentException(errorMessage);
  37.     }
  38.     return jaasConfig;
  39. }

4.跨集群访问开启Kerberos认证的Kafka

目的，Flink任务运行在开启Kerberos认证的A集群，同时读取A集群以及B集群的Kafka信息。

步骤：

上传新集群的Krb5.conf文件，追加到原有的Krb5文件中。
单独配置sasl.jaas.config参数。

特别注意：

新上传的Krb5.conf文件中，domain_realm一定要做realms的映射，否则会使用default_realm。

  1. [realms]
  2.  PLS.COM = {
  3.   kdc = plscdh00:88
  4.   admin_server = plscdh00
  5.  }
  6. 将plscdh0-3 映射到PLS.COM
  7. [domain_realm]
  8.      .pls.com = PLS.COM
  9.      pls.com = PLS.COM
  10.      plscdh01  = PLS.COM  
  11.      plscdh02 = PLS.COM
  12.      plscdh03 = PLS.COM
  13.      plscdh00 = PLS.COM

将新上传的Krb5.conf文件中的domain_realm以及realms追加到现有的Krb5文件中。
通过使用sasl.jaas.config来传递Jaas文件内容，如果使用AppConfigurationEntry类传递的话，kafka默认LoginContextName为KafkaClient,多个kafka集群下取出的AppConfigurationEntry会混乱。
相关代码：

krb5文件构建代码：

  1. public class KrbConfManager {
  2.     Logger LOG = LoggerFactory.getLogger(KrbConfManager.class);
  4.     private static final String JAVA_SECURITY_KRB5_CONF = "java.security.krb5.conf";
  6.     private KrbConfManager(){}
  8.     public void appendKrbConf(String krbConf) {
  9.         this.appendKrbConf(krbConf, System.getProperty("user.dir"));
  10.     }
  12.     /**
  13.      *  将新增的kbr5.conf文件，使用Config类方法进行解析后，追加到现有的kbr5.conf文件中
  14.      * @param krbConf
  15.      */
  16.     public void appendKrbConf(String krbConf, String directory) {
  17.         String krbConfPath = DtFileUtils.getFileAbsolutePath(krbConf);
  18.         LOG.info("krb conf abs path is {}", krbConfPath);
  19.         Preconditions.checkArgument(DtFileUtils.fileExistCheck(krbConfPath),"krb file does not exist");
  20.         try {
  22.             Constructor<Config> constructor = Config.class.getDeclaredConstructor();
  23.             constructor.setAccessible(true);
  24.             Config configParser = constructor.newInstance();
  26.             Method loadConfigFile = configParser.getClass().getDeclaredMethod("loadConfigFile", String.class);
  27.             loadConfigFile.setAccessible(true);
  28.             List<String> configFileList = (List<String>) loadConfigFile.invoke(configParser, krbConfPath);
  30.             Method parseStanzaTable = configParser.getClass().getDeclaredMethod("parseStanzaTable", List.class);
  31.             parseStanzaTable.setAccessible(true);
  33.             Hashtable<String, Object> appendConfig = (Hashtable<String, Object>) parseStanzaTable.invoke(configParser, configFileList);
  34.             Hashtable<String, Object> appendRealms = (Hashtable<String, Object>) appendConfig.get("realms");
  35.             Hashtable<String, Object> appendDomainRealm = (Hashtable<String, Object>) appendConfig.get("domain_realm");
  37.             Config instance = Config.getInstance();
  38.             Field stanzaTable = instance.getClass().getDeclaredField("stanzaTable");
  39.             stanzaTable.setAccessible(true);
  40.             Hashtable<String, Object> currentTable = (Hashtable<String, Object>) stanzaTable.get(instance);
  42.             Hashtable<String, Object> realms = (Hashtable<String, Object>) currentTable.computeIfAbsent("realms", key -> new Hashtable());
  43.             realms.putAll(appendRealms);
  45.             Hashtable<String, Object> domainRealm = (Hashtable<String, Object>) currentTable.computeIfAbsent("domain_realm", key -> new Hashtable());
  46.             domainRealm.putAll(appendDomainRealm);
  48.             StringBuffer stringBuffer = new StringBuffer();
  49.             String newKerbConfigStr = buildKrbConfigStr(currentTable, stringBuffer);
  50.             LOG.info("====buildKerbConf======\n{}", newKerbConfigStr);
  52.             String krb5FilePath = DtFileUtils.createTempFile("krb-", ".conf", newKerbConfigStr, directory);
  53.             System.setProperty(JAVA_SECURITY_KRB5_CONF, krb5FilePath);
  55.             Config.refresh();
  57.         } catch (Exception e) {
  58.             throw new RuntimeException("build krb conf error", e);
  59.         }
  60.     }
  62.     private String buildKrbConfigStr(Hashtable<String, Object> currentTable, StringBuffer stringBuilder) {
  63.         Set<String> keySet = currentTable.keySet();
  64.         for (String key : keySet) {
  65.             stringBuilder = stringBuilder.append("[").append(key).append("]").append("\n");
  66.             if (!StringUtils.equalsIgnoreCase(key, "realms")) {
  67.                 toStringInternal("", currentTable.get(key), stringBuilder);
  68.             } else {
  69.                 dealRealms(currentTable.get(key), stringBuilder);
  70.             }
  71.         }
  72.         return stringBuilder.toString();
  73.     }
  75.     private void dealRealms(Object realms, StringBuffer stringBuilder) {
  76.         if (realms instanceof Hashtable) {
  77.             Hashtable realmsTable = (Hashtable) realms;
  78.             Iterator tabIterator = realmsTable.keySet().iterator();
  80.             while (tabIterator.hasNext()) {
  81.                 Object entity = tabIterator.next();
  82.                 stringBuilder = stringBuilder.append(entity).append(" = ").append("{\n");
  83.                 toStringInternal("", realmsTable.get(entity), stringBuilder);
  84.                 stringBuilder.append("}\n");
  85.             }
  86.         }
  87.     }
  89.     private static void toStringInternal(String prefix, Object obj, StringBuffer sb) {
  90.         if (obj instanceof String) {
  91.             // A string value, just print it
  92.             sb.append(obj).append('\n');
  93.         } else if (obj instanceof Hashtable) {
  94.             // A table, start a new sub-section...
  95.             Hashtable<?, ?> tab = (Hashtable<?, ?>) obj;
  96.             for (Object o : tab.keySet()) {
  97.                 // ...indent, print "key = ", and
  98.                 sb.append(prefix).append("    ").append(o).append(" = ");
  99.                 // ...go recursively into value
  100.                 toStringInternal(prefix + "    ", tab.get(o), sb);
  101.             }
  102.             sb.append(prefix).append("\n");
  103.         } else if (obj instanceof Vector) {
  104.             // A vector of strings, print them inside [ and ]
  105.             Vector<?> v = (Vector<?>) obj;
  106.             boolean first = true;
  107.             for (Object o : v.toArray()) {
  108.                 if (!first) {
  109.                     sb.append(",");
  110.                 }
  111.                 sb.append(o);
  112.                 first = false;
  113.             }
  114.             sb.append("\n");
  115.         }
  116.     }
  118.     private enum Singleton {
  119.         INSTANCE;
  121.         private final KrbConfManager instance;
  123.         Singleton() {
  124.             instance = new KrbConfManager();
  125.         }
  127.         private KrbConfManager getInstance() {
  128.             return instance;
  129.         }
  130.     }
  132.     public static KrbConfManager getInstance() {
  133.         return Singleton.INSTANCE.getInstance();
  134.     }
  136. }

sasl.jaas.config 文件构建以及填充

  1. public interface SecurityManager {
  2.     Logger LOG = LoggerFactory.getLogger(SecurityManager.class);
  4.     String SASL_JAAS_CONFIG = "sasl.jaas.config";
  6.     default void kerberosSecurity(KafkaSourceTableInfo kafkaSourceTableInfo, Properties props) {
  7.         if (StringUtils.equalsIgnoreCase(kafkaSourceTableInfo.getKerberosAuthEnable(), Boolean.TRUE.toString())) {
  8.             Optional.ofNullable(kafkaSourceTableInfo.getKrbConfName())
  9.                     .ifPresent(KrbConfManager.getInstance()::appendKrbConf);
  11.             String jaasContent = JaasConfigUtil.JaasConfig.builder()
  12.                     .setLoginModule("com.sun.security.auth.module.Krb5LoginModule")
  13.                     .setLoginModuleFlag("required")
  14.                     .setPrincipal(DtStringUtil.addQuoteForStr(kafkaSourceTableInfo.getPrincipal()))
  15.                     .setKeyTab(DtStringUtil.addQuoteForStr(DtFileUtils.getFileAbsolutePath(kafkaSourceTableInfo.getKeyTab())))
  16.                     .setUseKeyTab(kafkaSourceTableInfo.getUseKeyTab())
  17.                     .setUseTicketCache(kafkaSourceTableInfo.getUseTicketCache())
  18.                     .build()
  19.                     .generateJaasConfigStr();
  20.             LOG.info(" kafka jaas Content: \n{}", jaasContent);
  22.             props.put(SASL_JAAS_CONFIG, jaasContent);
  23.         }
  24.     }
  26. }

构建入口，要在OPEN中调用

  1. KafkaConsumer010#open
  2. @Override
  3.     public void open(Configuration configuration) throws Exception {
  4.         kerberosSecurity(kafkaSourceTableInfo, this.properties);
  5.         super.open(configuration);
  6.     }

版权说明 : 本文为转载文章, 版权归原作者所有 版权申明

原文链接 : https://blog.csdn.net/qq_21383435/article/details/126070842

内容来源于网络，如有侵权，请联系作者删除！

hadoopflinkkafka

相关文章

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com