Managing Messages in Kafka Topics

Scenario

You can produce or consume messages in Kafka topics using the MRS cluster client. For clusters with Kerberos authentication enabled, you must have the permission to perform these operations.

Prerequisites

You have installed the Kafka client.

Procedure

1. Go to the Kafka service page.

   Log in to FusionInsight Manager and choose Cluster > Name of the desired cluster > Services > Kafka.

2. Click instance. Query the IP addresses of the Kafka instances.

   Record the IP address of any Kafka instance.

3. Prepare the client based on service requirements. Log in to the node where the client is installed.

   Log in to the node where the client is installed. For details, see Installing a Client.

4. Run the following command to switch to the client directory, for example, /opt/client/Kafka/kafka/bin.

   cd /opt/client/Kafka/kafka/bin

5. Run the following command to configure environment variables:

   source /opt/client/bigdata_env

6. For clusters with Kerberos authentication enabled, run the following command to authenticate the user. For clusters with Kerberos authentication disabled, skip this step.

   kinit Kafka user

   Example:

   kinit admin

7. Manage messages in Kafka topics using the following commands:
   • Producing messages

     sh kafka-console-producer.sh --broker-list IP address of the node where the Kafka instance resides:9092 --topic Topic name --producer.config /opt/client/Kafka/kafka/config/producer.properties

     You can input specified information as the messages produced by the producer and then press Enter to send the messages. To end message producing, press Ctrl + C to exit.

   • Consuming messages

     sh kafka-console-consumer.sh --topic Topic name --bootstrap-server IP address of the node where the Kafka instance resides:9092 --consumer.config /opt/client/Kafka/kafka/config/consumer.properties

     In the configuration file, group.id (indicating the consumer group) is set to example-group1 by default. Users can change the value as required. The value takes effect each time consumption occurs.

     By default, the system reads unprocessed messages in the current consumer group when the command is executed. If a new consumer group is specified in the configuration file and the --from-beginning parameter is added to the command, the system reads all messages that have not been automatically deleted in Kafka.

   

   • For the IP address of the node where the Kafka instance locates, use the IP address of any broker instance.
   • If Kerberos authentication is enabled, change the port to 21007.
   • By default, the ZooKeeper's clientPort value is 2181.