Hello, everyone!
This post describes Data Consumption Fails and the Consumer Stays in the Waiting State.
[Applicable Versions]
6.5.x
[Symptom]
The cluster of the FusionInsight V100R002C60 version is installed. ZooKeeper and Kafka are installed.
When the Consumer consumes data from Kafka, the client stays in the Waiting state.
[Possible Causes]
The Kafka service is abnormal.
The Consumer client adopts non-security access and access is disabled on the server.
The Consumer client adopts non-security access and ACL is set for Kafka topics.
[Fault Locating]
The possible reasons why the Consumer fails to consume data from Kafka may be related to the Consumer or Kafka.
On the FusionInsight Manager page, choose Services > Kafka Status. The status of the Kafka cluster is Good and the monitoring indicators show no errors.
Check the Consumer client log. It is found that the information about the frequent connections and disconnections to the Broker node is printed, as shown in the following output.
[root@10-10-144-2 client]# kafka-console-consumer.sh --topic test --zookeeper 10.5.144.2:24002/kafka --from-beginning
[2017-03-07 01:22,658] INFO Fetching metadata from broker BrokerEndPoint(1,10.5.144.2,21005) with correlation id 26 for 1 topic(s) Set(test) (kafka.client.ClientUtils$)
[2017-03-07 01:22,659] INFO Connected to 10.5.X.X:21005 for producing (kafka.producer.SyncProducer)
[2017-03-07 01:22,659] INFO Disconnecting from 10.5.X.X:21005 (kafka.producer.SyncProducer)
The Consumer accesses Kafka using port 21005, which is a non-security port.
On the FusionInsight Manager page, choose Services > Kafka Configuration > Configuration to check the current Kafka cluster configurations. It is found that allow.everyone.if.no.acl.found is set to false.
If ACL is set to false, port 21005 cannot be used for access. For details, see section 13.10.2.5.1-Safety Instruction on Using Kafka.
Check the Consumer client log. It is found that the information about the frequent connections and disconnections to the Broker node is printed, as shown in the following output.
[root@10-10-144-2 client]# kafka-console-consumer.sh --topic test_acl --zookeeper 10.5.X.X:24002/kafka --from-beginning
[2017-03-07 01:22,992] INFO Fetching metadata from broker BrokerEndPoint(2,10.5.X.X,21005) with correlation id 16 for 1 topic(s) Set(topic_acl) (kafka.client.ClientUtils$)
[2017-03-07 01:22,993] INFO Connected to 10.5.X.X:21005 for producing (kafka.producer.SyncProducer)
[2017-03-07 01:22,994] INFO Disconnecting from 10.5.X.X:21005 (kafka.producer.SyncProducer)
The Consumer accesses Kafka using port 21005, which is a non-security port.
Run the client command to check the ACL permission of the topic.
[root@10-10-X-X client]# kafka-acls.sh --authorizer-properties zookeeper.connect=10.5.X.X:24002/kafka --list --topic topic_aclCurrent ACLs for resource `Topic:topic_acl`:User:test_user has Allow permission for operations: Describe from hosts: *User:test_user has Allow permission for operations: Write from hosts: *
If ACL is set for the topic, port 21005 cannot be used for access. For details, see section 13.10.2.5.1-Safety Instruction on Using Kafka.
The following information is printed in the Consumer client log:
[root@10-10-X-X client]# kafka-console-consumer.sh --topic topic_acl --bootstrap-server 10.5.X.X:21007 --consumer.config /opt/client/Kafka/kafka/config/consumer.properties --from-beginning --new-consumer
[2017-03-07 01:22,478] INFO Kafka version : 0.9.0.0 (org.apache.kafka.common.utils.AppInfoParser)
[2017-03-07 01:22,478] INFO Kafka commitId : unknown (org.apache.kafka.common.utils.AppInfoParser)
The Consumer uses port 21007 to access Kafka.
Run the klist client command to query the current authenticated user.
[root@10-10-X-X client]# klist
Ticket cache: FILE:/tmp/krb5cc_0
Default principal: test@HADOOP.COM
Valid starting Expires Service principal
01/25/17 11:06:48 01/26/17 11:06:45 krbtgt/HADOOP.COM@HADOOP.COM
The current authenticated user is test.
Run the client command to check the ACL permission of the topic.
[root@10-10-X-X client]# kafka-acls.sh --authorizer-properties zookeeper.connect=10.5.X.X:24002/kafka --list --topic topic_aclCurrent ACLs for resource `Topic:topic_acl`:User:test_user has Allow permission for operations: Describe from hosts: *User:test_user has Allow permission for operations: Write from hosts: *User:ttest_user has Allow permission for operations: Read from hosts: *
If ACL is set for the topic, user testdoes not have the permission to perform the consumer operation.
Repeat 2 to solve it.
Log in to Kafka Broker using SSH.
Run the cd /var/log/Bigdata/kafka/broker command to go to the log directory.
Check the kafka-authorizer.log log. It shows that the user does not belong to the kafkaor kafkaadmin group.
2017-03-07 01:22,648 | INFO | [kafka-request-handler-0] | The principal is test, belongs to Group : [hadoop, ficommon] | kafka.authorizer.logger (SimpleAclAuthorizer.scala:169)2017-03-07 01:22,648 | WARN | [kafka-request-handler-0] | The user is not belongs to kafka or kafkaadmin group, authorize failed! | kafka.authorizer.logger (SimpleAclAuthorizer.scala:170)
Repeat 3 to solve it.
[Solution]
Set allow.everyone.if.no.acl.found to true and restart the Kafka service.
Sse the account with login permission.
For example:
kinit test_user
Alternatively, grant the user with related permissions.
../Resource/public_sys-resources/notice_en.png?v=20200527
A Kafka administrator user who belongs to the kafkaadmin group is required for performing the operations.
Example:
kafka-acls.sh --authorizer-properties zookeeper.connect=10.5.144.2:24002/kafka --topic topic_acl --consumer --add --allow-principal User:test --group test
[root@10-10-X-X client]# kafka-acls.sh --authorizer-properties zookeeper.connect=8.5.X.X:24002/kafka --list --topic topic_acl
Current ACLs for resource `Topic:topic_acl`:
User:test_user has Allow permission for operations: Describe from hosts: *
User:test_user has Allow permission for operations: Write from hosts: *
User:test has Allow permission for operations: Describe from hosts: *
User:test has Allow permission for operations: Write from hosts: *
User:test has Allow permission for operations: Read from hosts: *
Add the user to the kafka or kafkaadmin group.
Hope you can learn from it, thank you!
