Hi Kafka experts,

I have enabled KAFKA 2.2.x parcel (kafka version 0.10.2) in CDH 5.12. When I run a basic producer or consumer command such as:

[root@~]# /opt/cloudera/parcels/KAFKA-2.2.0-1.2.2.0.p0.68/lib/kafka/bin/kafka-console-producer.sh --broker-list xyz1.com:9092 xyz2.com:9092 --topic topic1
SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/opt/cloudera/parcels/KAFKA-2.2.0-1.2.2.0.p0.68/lib/kafka/libs/slf4j-log4j12-1.7.21.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/opt/cloudera/parcels/KAFKA-2.2.0-1.2.2.0.p0.68/lib/kafka/libs/slf4j-log4j12-1.7.5.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
17/12/21 12:54:21 INFO producer.ProducerConfig: ProducerConfig values:
acks = 1
batch.size = 16384
block.on.buffer.full = false

....

ssl.truststore.location = null
ssl.truststore.password = null
ssl.truststore.type = JKS
timeout.ms = 30000
value.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer

17/12/21 12:54:21 INFO utils.AppInfoParser: Kafka version : 0.10.2-kafka-2.2.0
17/12/21 12:54:21 INFO utils.AppInfoParser: Kafka commitId : unknown

hello hello

17/12/21 12:56:26 WARN clients.NetworkClient: Error while fetching metadata with correlation id 1 : {topic1=UNKNOWN_TOPIC_OR_PARTITION}
17/12/21 12:56:27 WARN clients.NetworkClient: Error while fetching metadata with correlation id 2 : {topic1=UNKNOWN_TOPIC_OR_PARTITION}
17/12/21 12:56:27 WARN clients.NetworkClient: Error while fetching metadata with correlation id 3 : {topic1=UNKNOWN_TOPIC_OR_PARTITION}

This CDH cluster has Sentry enabled but no Kerberos and no SSL. I think there is a permission issue for the user as I get the below in the /var/log/kafka/kafka-broker-xyz.log

2017-12-21 13:00:18,813 WARN org.apache.sentry.provider.common.HadoopGroupMappingService: Unable to obtain groups for ANONYMOUS
java.io.IOException: No groups found for user ANONYMOUS
at org.apache.hadoop.security.Groups.noGroupsForUser(Groups.java:199)
at org.apache.hadoop.security.Groups.getGroups(Groups.java:222)
at org.apache.sentry.provider.common.HadoopGroupMappingService.getGroups(HadoopGroupMappingService.java:60)
at org.apache.sentry.provider.common.ResourceAuthorizationProvider.getGroups(ResourceAuthorizationProvider.java:167)
at org.apache.sentry.provider.common.ResourceAuthorizationProvider.doHasAccess(ResourceAuthorizationProvider.java:97)
at org.apache.sentry.provider.common.ResourceAuthorizationProvider.hasAccess(ResourceAuthorizationProvider.java:91)
at org.apache.sentry.kafka.binding.KafkaAuthBinding.authorize(KafkaAuthBinding.java:212)
at org.apache.sentry.kafka.authorizer.SentryKafkaAuthorizer.authorize(SentryKafkaAuthorizer.java:63)
at kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$authorize$1.apply(KafkaApis.scala:343)
at kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$authorize$1.apply(KafkaApis.scala:343)
at scala.Option.forall(Option.scala:247)
at kafka.server.KafkaApis.kafka$server$KafkaApis$$authorize(KafkaApis.scala:343)
at kafka.server.KafkaApis$$anonfun$39.apply(KafkaApis.scala:838)
at kafka.server.KafkaApis$$anonfun$39.apply(KafkaApis.scala:838)
at scala.collection.TraversableLike$$anonfun$partition$1.apply(TraversableLike.scala:314)
at scala.collection.TraversableLike$$anonfun$partition$1.apply(TraversableLike.scala:314)
at scala.collection.immutable.Set$Set1.foreach(Set.scala:94)
at scala.collection.TraversableLike$class.partition(TraversableLike.scala:314)
at scala.collection.AbstractTraversable.partition(Traversable.scala:104)
at kafka.server.KafkaApis.handleTopicMetadataRequest(KafkaApis.scala:838)
at kafka.server.KafkaApis.handle(KafkaApis.scala:83)
at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:62)
at java.lang.Thread.run(Thread.java:745)
2017-12-21 13:00:19,067 WARN org.apache.sentry.provider.common.HadoopGroupMappingService: Unable to obtain groups for ANONYMOUS

What is the correct way to setup the Sentry authorization to give permission to the user on kafka? Any blog or instructions will be greatly appreciated. Thanks!