private Token<?> getKafkaDelegationTokenForBrokers(Configuration conf, TableDesc tableDesc) {
    Properties tableProperties = tableDesc.getProperties();
    String kafkaBrokers = (String) tableProperties.get(HIVE_KAFKA_BOOTSTRAP_SERVERS.getName());
    LOG.info("Getting kafka credentials for brokers: {}", kafkaBrokers);

    String keytab = HiveConf.getVar(conf, HiveConf.ConfVars.HIVE_SERVER2_KERBEROS_KEYTAB);
    String principal = HiveConf.getVar(conf, HiveConf.ConfVars.HIVE_SERVER2_KERBEROS_PRINCIPAL);
    try {
      principal = SecurityUtil.getServerPrincipal(principal, "0.0.0.0");
    } catch (IOException e) {
      throw new RuntimeException(e);
    }

    Properties config = new Properties();
    config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBrokers);
    SecurityProtocol protocol = KafkaUtils.securityProtocol(tableProperties);
    if (protocol == null) {
      protocol = SecurityProtocol.SASL_PLAINTEXT;
      LOG.warn("Kafka security.protocol is undefined in table properties. Using default {}", protocol.name);
    }
    config.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, protocol.name);

    String jaasConfig =
        String.format("%s %s %s %s serviceName=\"%s\" keyTab=\"%s\" principal=\"%s\";",
            "com.sun.security.auth.module.Krb5LoginModule required", "debug=true", "useKeyTab=true",
            "storeKey=true", "kafka", keytab, principal);
    config.put(SaslConfigs.SASL_JAAS_CONFIG, jaasConfig);

    LOG.debug("Jaas config for requesting kafka credentials: {}", jaasConfig);
    Configuration confCopy = new Configuration(conf);
    tableProperties.stringPropertyNames().forEach(key -> confCopy.set(key, tableProperties.getProperty(key)));
    KafkaUtils.setupKafkaSslProperties(confCopy, config);
    CreateDelegationTokenOptions createDelegationTokenOptions = new CreateDelegationTokenOptions();
    try (AdminClient admin = AdminClient.create(config)) {
      CreateDelegationTokenResult createResult = admin.createDelegationToken(createDelegationTokenOptions);
      DelegationToken token = createResult.delegationToken().get();
      LOG.info("Got kafka delegation token: {}", token);
      return new Token<>(token.tokenInfo().tokenId().getBytes(), token.hmac(), null, new Text("kafka"));
    } catch (ExecutionException e) {
      throw new RuntimeException("Exception while getting Kafka token", e);
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
      throw new RuntimeException("Interrupted while getting Kafka token", e);
    }
  }
Returns whether a Kafka token is required for performing operations on the specified table.
If "security.protocol" is set to "PLAINTEXT", we don't need to collect delegation token at all.

@return true if a Kafka token is required for performing operations on the specified table and false otherwise.
kafka-handler/src/java/org/apache/hadoop/hive/kafka/KafkaDagCredentialSupplier.java