Broker replication error "Not authorized to access topics: [Topic authorization failed.] "

Question

I have three Kafka brokers (kafka_2.11-0.10.0.0)and each broker's security is configured as below,

listeners=PLAINTEXT://xxxx:9093,SASL_PLAINTEXT://xxxx:10093
advertised.listeners=PLAINTEXT://xxxx:9093,SASL_PLAINTEXT://xxxx:10093
security.inter.broker.protocol=PLAINTEXT
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
allow.everyone.if.no.acl.found=true
super.users=User:admin

kafka_server_jaas.conf also configured with admin user.

KafkaServer {
        org.apache.kafka.common.security.plain.PlainLoginModule required
        username="admin"
        password="welcome1"
        user_guest="welcome1";
};

When I connect use PLAINTEXT listener and produce and consume messages, everything works fine.

but when I try to add ACLs to some topic, the cluster will output error messages like below:

bin/kafka-acls.sh --authorizer-properties zookeeper.connect=xxxx:2181/kafka10  --add --allow-principal User:guest --producer --topic page_visits_10k  

Broker Output

[2016-05-31 10:49:57,497] ERROR [ReplicaFetcherThread-0-2], Error for partition [page_visits_10k,1] to broker 2:org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [Topic authorization failed.] (kafka.server.ReplicaFetcherThread)
[2016-05-31 10:49:59,003] ERROR [ReplicaFetcherThread-0-2], Error for partition [page_visits_10k,1] to broker 2:org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [Topic authorization failed.] (kafka.server.ReplicaFetcherThread)

Error output looks like inter broker replication error, appreciate for any help.

Answer 1

I have faced similar issue with using the ACLs in Kafka v.0.10. I found this discussion helpful. Especially enabling the authorization log in order to check what is the incoming username for the request and what is it specified in your ACLs.
Authorization log can be enabled by modifying the log4j.properties in the config folder. In log4j.properties file, change WARN to DEBUG and restart the kafka-servers.

log4j.logger.kafka.authorizer.logger=DEBUG, authorizerAppender

This helped me in sorting out my issue. Hope that helps.

PS: The authorization logs generated will be very lengthy and consume a lot of space. So, remember to turn this off when done with debugging.

Answer 2

Finally I figure out this issue.
The error is caused by:
security.inter.broker.protocol=PLAINTEXT

It should be:
security.inter.broker.protocol=SASL_PLAINTEXT

[2016-05-31 10:49:57,497] ERROR [ReplicaFetcherThread-0-2], Error for partition [page_visits_10k,1] to broker 2:org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [Topic authorization failed.] (kafka.server.ReplicaFetcherThread)
[2016-05-31 10:49:59,003] ERROR [ReplicaFetcherThread-0-2], Error for partition [page_visits_10k,1] to broker 2:org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [Topic authorization failed.] (kafka.server.ReplicaFetcherThread)