Managing Messages in Kafka Topics
Users can produce or consume messages in Kafka topics using the MRS cluster client. For clusters with Kerberos authentication enabled, the user must have the permission to perform these operations.
The client has been updated.
- On the MRS cluster details page, click Components.
- If the Components tab is not displayed on the cluster details page, complete IAM user synchronization first. (On the Dashboard tab page of the cluster details page, click on the right side of IAM User Sync to synchronize IAM users.)
- For MRS 1.8.10 or earlier, log in to MRS Manager. For details, see Accessing MRS Manager. Then, choose Services.
- Choose . Query the IP addresses of the Kafka instances.
Record the IP address of any Kafka instance.
- Log in to the node where the client is installed.
For example, if you have updated the client on the Master2 node, log in to the Master2 node to use the client. For details, see Using an MRS Client.
- Run the following command to switch the user:
sudo su - omm
- Run the following command to switch to the client directory, for example, /opt/client/Kafka/kafka/bin.
- Run the following command to configure the environment variables:
- If Kerberos authentication is enabled, run the following command to authenticate the user. If Kerberos authentication is disabled, skip this step.
kinit Kafka username
For example, kinit admin
- Manage messages in Kafka topics using the following commands:
- Produce messages.
sh kafka-console-producer.sh --broker-list IP address of the node where the Kafka instance is located: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.
- Consume messages.
sh kafka-console-consumer.sh --topic Topic name --bootstrap-server IP address of the node where the Kafka instance is located:9092 --new-consumer --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 a 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 is located, use the IP address of any Broker instance.
- If Kerberos authentication is enabled, change the port to 21007.
- Produce messages.