How to use two Kerberos keytabs (for Kafka and Hadoop HDFS) from a Flink job on a Flink standalone cluster?

Question

Question

On a Flink standalone cluster, running on a server, I am developing a Flink streaming job in Scala. The job consumes data from more than 1 Kafka topics, (do some formatting,) and write results to HDFS.

One of the Kafka topic, and HDFS, they both require separate Kerberos authentications (because they belong to completely different clusters).

My questions are:

1. Is it possible (if yes, how?) to use two Kerberos keytabs (one for Kafka, the other for HDFS) from a Flink job on a Flink cluster, running on a server? (so the Flink job can consume from Kafka topic and write to HDFS at the same time)
2. If not possible, what is a reasonable workaround, for the Kafka-Flink-HDFS data streaming when Kafka and HDFS are both Kerberos protected?

Note

• I am quite new to the most of the technologies mentioned here.
• The Flink job can write to HDFS if it doesn't need to consume the Kerberos-requiring topic. In this case, I specified the information of HDFS tosecurity.kerberos.login.keytab and security.kerberos.login.principal in flink-conf.yaml
• I am using HDFS Connector provided from Flink to write to HDFS.

• Manually switching the Kerberos authentication between the two principals was possible. In [realm] section in krb5.conf file, I specified two realms, one for Kafka, the other for HDFS.

  kinit -kt path/to/hdfs.keytab [principal: [email protected]...]

  kinit -kt path/to/kafka.keytab [principal: [email protected]...]

Environment

• Flink (v1.4.2) https://ci.apache.org/projects/flink/flink-docs-stable/
• Kafka client (v0.10.X)
• HDFS (Hadoop cluster HDP 2.6.X)

Thanks for your attentions and feedbacks!

Kinghorn

"separate Kerberos authentications (because they belong to completely different clusters)" > that's not a very convincing argument. The same Kerberos TGT (proof of identity) can be granted Kerberos service tickets in different target realms, if there is "trust" between realms (e.g. dedicated MIT Kerberos realm for cluster services vs global Active Directory forest for auth), or if both realms are attached to the same root realm (e.g. different AD domains in the same forest). That's how it works in a typical Big Corp environment with tight security. – Entropy

But even with a single Kerberos principal/keytab, you might run into problems because in general (TBC with Flink) the HDFS client uses the static Hadoop UserGroupInformation which bypasses part of the standard Java implementation of Kerberos, and in general (...) the Kafka client uses raw JAAS configuration for standard Java impl. They don't play well together. Or at least, they did not when I tested HDFS + Hive JDBC in custom Java code a few years ago ; maybe Flink bypasses the UGI to avoid side effects. – Entropy

I face the same problem.We are currently using Hadoop Map-Reduce to handle this situation.Start the child process in the map, use the main process to read the data from kafka and send it to the child process, which then sends the data to HDFS. – Cremona

Answer 1

After three years from my initial post, our architecture has moved from standalone bare metal server to Docker container on Mesos, but let me summarize the workaround (for Flink 1.8):

• Place krb5.conf with all realm definitions and domain-realm mappings (for example under /etc/ of the container)

• Place Hadoop krb5.keytab (for example under /kerberos/HADOOP_CLUSTER.ORG.EXAMPLE.COM/)

• Configure Flink's security.kerberos.login.* properties in flink-conf.yaml

  • security.kerberos.login.use-ticket-cache: true
  • security.kerberos.login.principal: username@HADOOP_CLUSTER.ORG.EXAMPLE.COM
  • security.kerberos.login.contexts should not be configured. This ensures that Flink does not use Hadoop’s credentials for Kafka and Zookeeper.

• Copy keytabs for Kafka into separate directories inside the container (for example under /kerberos/KAFKA_CLUSTER.ORG.EXAMPLE.COM/)

• Periodically run custom script to renew ticket cache

KINIT_COMMAND_1='kinit -kt /kerberos/HADOOP_CLUSTER.ORG.EXAMPLE.COM/krb5.keytab username@HADOOP_CLUSTER.ORG.EXAMPLE.COM'
KINIT_COMMAND_2='kinit -kt /kerberos/KAFKA_CLUSTER.ORG.EXAMPLE.COM/krb5.keytab username@KAFKA_CLUSTER.ORG.EXAMPLE.COM -c /tmp/krb5cc_kafka'
...

• Set the property sasl.jaas.config when instantiating each FlinkKafkaConsumer to the actual JAAS configuration string.
  • To bypass the global JAAS configuration. If we set this globally, we can’t use different Kafka instances with different credentials, or unsecured Kafka together with secured Kafka.

props.setProperty("sasl.jaas.config", 
    "com.sun.security.auth.module.Krb5LoginModule required " +
    "refreshKrb5Config=true " +
    "useKeyTab=true " +
    "storeKey=true " +
    "debug=true " +
    "keyTab=\"/kerberos/KAFKA_CLUSTER.ORG.EXAMPLE.COM/krb5.keytab\" " +
    "principal=\"username@KAFKA_CLUSTER.ORG.EXAMPLE.COM\";")

Kinghorn

Answer 2

Based on the answer and comment on this very similar question

It appears there is no clear way to use two credentials in a single Flink job.

Promising approaches or workarounds:

• Creating a trust
• Co-installing Kafka and HDFS on the same platform
• Using something else to bridge the gap

An example of the last point:

You can use something like NiFi or Streams Replication Manager to bring the data from the source Kafka, to the Kafka in your cluster. NiFi is more modular, and it is possible to configure the kerberos credentials for each step. Afterward you are in a single context which Flink can handle.

Full disclosure: I am an employee of Cloudera, a driving force behind NiFi, Kafka, HDFS, Streams Replication Manager and since recently Flink

Clarkin