Accessing Kerberos enabled Kafka topics using GetKafka/PutKafka Processor

[#1]

With HDF 1.1.2.1 release, HDF supports accessing Kerberos enabled kafka topics. For a stanalone NiFi node, following instructions can be used.

1. new jass file should be created with following entries.

Client {
  com.sun.security.auth.module.Krb5LoginModule required
  useKeyTab=true
  keyTab="./conf/nifi.keytab"
  useTicketCache=false
  principal="nifi@EXAMPLE.COM”; 
};
KafkaClient {
   com.sun.security.auth.module.Krb5LoginModule required
   useTicketCache=false
   renewTicket=true
   serviceName="kafka"
   useKeyTab=true
   keyTab="./conf/nifi.keytab"
   principal="nifi@EXAMPLE.COM";
};

*** Both the KafkaClient and Client configs should use the same principal and key tab ***

2. bootstrap.conf file should be changed to include following line.

java.arg.15=-Djava.security.auth.login.config=/<path>/zookeeper-jaas.conf

Restart NiFi node after changes to bootstrap.conf.

3. GetKafka/PutKafka processors should be modified to include a new property security.protocol and value for this property should be set to PLAINTEXTSASL

kafka.png

That's it ! Now NiFi can read from Kerberos enabled kafka topics.

Thank You @rgarcia for helping me with these configurations.

[#2]

There are additional items that will need to be taken in to consideration if you are running a NiFi cluster. See the following for more details: https://community.hortonworks.com/content/kbentry/28180/how-to-configure-hdf-12-to-send-to-and-get-d...

[#3]

Hi,

I have been trying to configure Nifi to work with kerberized Kafka, but to no avail. Here my zookeeper-jaas.conf:

Client {
com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
keyTab="./conf/user.keytab"
storeKey=true
useTicketCache=false
principal="user@REALM.COM”;
};
KafkaClient {
com.sun.security.auth.module.Krb5LoginModule required
useTicketCache=true
renewTicket=true
serviceName="kafka"
useKeyTab=true
keyTab="./conf/user.keytab"
principal="user@REALM.COM";
};

I am getting errors when starting the PutKafka processor:

Caused by: java.io.IOException: Configuration Error:
Line 8: expected [option key]
at sun.security.provider.ConfigFile$Spi.ioException(ConfigFile.java:666) ~[na:1.8.0_66]
at sun.security.provider.ConfigFile$Spi.match(ConfigFile.java:562) ~[na:1.8.0_66]
at sun.security.provider.ConfigFile$Spi.parseLoginEntry(ConfigFile.java:477) ~[na:1.8.0_66]
at sun.security.provider.ConfigFile$Spi.readConfig(ConfigFile.java:427) ~[na:1.8.0_66]
at sun.security.provider.ConfigFile$Spi.init(ConfigFile.java:329) ~[na:1.8.0_66]
at sun.security.provider.ConfigFile$Spi.init(ConfigFile.java:227) ~[na:1.8.0_66]
at sun.security.provider.ConfigFile$Spi.access$000(ConfigFile.java:115) ~[na:1.8.0_66]
at sun.security.provider.ConfigFile$Spi$1.run(ConfigFile.java:180) ~[na:1.8.0_66]
at sun.security.provider.ConfigFile$Spi$1.run(ConfigFile.java:169) ~[na:1.8.0_66]
at java.security.AccessController.doPrivileged(Native Method) ~[na:1.8.0_66]
at sun.security.provider.ConfigFile$Spi.<init>(ConfigFile.java:169) ~[na:1.8.0_66]
at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method) ~[na:1.8.0_66]
at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62) ~[na:1.8.0_66]
at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45) ~[na:1.8.0_66]
at java.lang.reflect.Constructor.newInstance(Constructor.java:422) ~[na:1.8.0_66]
at java.security.Provider$Service.newInstance(Provider.java:1609) ~[na:1.8.0_66]
... 26 common frames omitted

I have checked the zookeeper-jaas.conf many times and cannot identify the error. Any pointers will be appreciated.

[#4]

Looks like it was a syntactical error in the file, not sure what exactly. By recreating the contents it works, which seems like a whitespace or some other character to be the reason for the error.

[#5]

I did the changes as specified in the above link(https://community.hortonworks.com/articles/28180/how-to-configure-hdf-12-to-send-to-and-get-data-fr.html) and configured getkafka processor. But I am getting below error when I start the getKafka processor.

********************Error Log*********

2017-08-24 10:10:19,964 INFO [test-consumer-group_storagev4-1c-1503583093158-c73a2251-leader-finder-thread] kafka.consumer.ConsumerFetcherManager [ConsumerFetcherManager-1503583093205] Added fetcher for partitions ArrayBuffer() 2017-08-24 10:10:20,169 WARN [test-consumer-group_storagev4-1c-1503583093158-c73a2251-leader-finder-thread] k.c.ConsumerFetcherManager$LeaderFinderThread [test-consumer-group_storagev4-1c-1503583093158-c73a2251-leader-finder-thread], Failed to find leader for Set([indexing,0]) java.lang.NullPointerException: null

************************

[#6]

Attached are the logs that are related to my getKafka processor. Can someone please go through the same and let me know what might be going wrong!getkafka-processor-log.txt