Search code examples
hadoopapache-kafkahdfsapache-flinkflink-streaming

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

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: xxx@XXX.XXX...]

    kinit -kt path/to/kafka.keytab [principal: yyy@YYY.YYY...]

Environment

Thanks for your attentions and feedbacks!

Solution

  • 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\";")