Using the Kafka Client

Scenario

This section guides users to use a Kafka client in an O&M or service scenario.

This section applies to MRS 3.x or later clusters.

Prerequisites

• The client has been installed. For example, the installation directory is /opt/client.
• Service component users have been created by the MRS cluster administrator. Machine-machine users need to download the keytab file. A human-machine user must change the password upon the first login. (Not involved in normal mode)
• After changing the domain name of a cluster, redownload the client to ensure that the kerberos.domain.name value in the configuration file of the client is set to the correct server domain name.

Using the Kafka Client

1. Log in to the node where the client is installed as the client installation user.
2. Run the following command to go to the client installation directory:

   cd /opt/client

3. Run the following command to configure environment variables:

   source bigdata_env

4. Run the following command to perform user authentication (skip this step in normal mode):

   kinit Component service user

5. Run the following command to switch to the Kafka client installation directory:

   cd Kafka/kafka/bin

6. Run the following command to use the client tool to view and use the help information:
   • ./kafka-console-consumer.sh: Kafka message reading tool
   • ./kafka-console-producer.sh: Kafka message publishing tool
   • ./kafka-topics.sh: Kafka topic management tool

7. If you need to use kafka-topics.sh to manage Kafka topics, run the following command:
   

   • Service IP address of the ZooKeeper node: Log in to FusionInsight Manager, choose Cluster > Services > ZooKeeper > Instance, and check and record the service IP address of the ZooKeeper role instance.
   • clientPort: You can search for clientPort in all ZooKeeper configuration parameters. The default ports are as follows:

     The default open-source port number is 2181.

     The default customized port number is 24002.

     Port customization/open source: When creating an LTS version cluster, you can set Component Port to Open source or Custom. If Open source is selected, the open source port is used. If Custom is selected, the customized port is used.

   • Kafka cluster IP address: Log in to FusionInsight Manager, choose Cluster > Services > Kafka > Instance, and view and record the service IP address of any broker instance.
   • The default IP address and port number of the Kafka cluster are 21007 in security mode and 9092 in common mode.
   • In MRS 3.3.1 and later versions, Kafka uses --bootstrap-server instead of --zookeeper to create topics.
   • The differences between --zookeeper and --bootstrap-server are as follows:
     • In --zookeeper mode, the client generates a copy allocation scheme. The community supports this mode from the beginning. To reduce the dependency on the ZooKeeper component, the community will delete the support for this mode in later versions. When creating a topic in this mode, you can select a copy allocation policy by combining the --enable-rack-aware and --enable-az-aware options. Note: The --enable-az-aware option can be used only when the cross-AZ feature is enabled on the server, that is, az.aware.enable is set to true. Otherwise, the execution fails.
     • In --bootstrap-server mode, the server generates a copy allocation solution. In later versions, the community supports only this mode for topic management. When a topic is created in this mode, the --enable-rack-aware and --enable-az-aware options cannot be used to control the copy allocation policy. The rack.aware.enable and az.aware.enable parameters can be used together to control the copy allocation policy. Note that the az.aware.enable parameter cannot be modified; if the cross-AZ feature is enabled during cluster creation, this parameter is automatically set to true; the rack.aware.enable parameter can be customized.
   • Creating a topic:

     ./kafka-topics.sh --create --topic topic name --partitions number of partitions occupied by the topic --replication-factor number of replicas of the topic --zookeeper Service IP address of the ZooKeeper node:clientPort/kafka

     ./kafka-topics.sh --create --topic topic name --partitions number of partitions occupied by the topic --replication-factor number of replicas of the topic --bootstrap-server IP address of the Kafka cluster:21007 --command-config ../config/client.properties

   • List of topics:
     • ./kafka-topics.sh --list --zookeeper Service IP address of the ZooKeeper node:clientPort/kafka
     • ./kafka-topics.sh --list --bootstrap-server IP address of the Kafka cluster:21007 --command-config ../config/client.properties
   • Viewing the topic:
     • ./kafka-topics.sh --describe --zookeeper Service IP address of the ZooKeeper node:clientPort/kafka --topic topic name
     • ./kafka-topics.sh --describe --bootstrap-server IP address of the Kafka cluster:21007 --command-config ../config/client.properties --topic topic name
   • Modifying a topic:
     • ./kafka-topics.sh --alter --topic topic name --config Configuration item=Configuration item --zookeeper Service IP address of the ZooKeeper node:clientPort/kafka
   • Expanding partitions:
     • ./kafka-topics.sh --alter --topic topic name --zookeeper Service IP address of the ZooKeeper node:clientPort/kafka --command-config Kafka/kafka/config/client.properties --partitions Number of partitions after the expansion
     • ./kafka-topics.sh --alter --topic topic name --bootstrap-server IP address of the Kafka cluster:21007 --command-config Kafka/kafka/config/client.properties --partitions Number of partitions after the expansion
   • Deleting a topic:
     • ./kafka-topics.sh --delete --topic topic name --zookeeper Service IP address of the ZooKeeper node:clientPort/kafka
     • ./kafka-topics.sh --delete --topic topic name --bootstrap-server IP address of the Kafka cluster:21007 --command-config ../config/client.properties