Step 4: Connect to a Kafka Instance to Create and Retrieve Messages

Prerequisites

• You have correctly configured security group rules. For details, see Table 1.
• The instance connection address has been obtained.
  Figure 1 Kafka instance (SASL enabled) connection addresses for intra-VPC access
  
• You have obtained the name of the topic created in (Optional) Step 3: Create a Topic.
• You have purchased an ECS, installed the JDK, configured the environment variables, and downloaded a Kafka client. For details, see Step 1: Prepare the Environment.

Preparing the Configuration File for Message Creation and Retrieval

1. Log in to a Linux ECS.
2. Map hosts to IP addresses in the /etc/hosts file on the ECS, so that the client can quickly parse the instance brokers.

   Set IP addresses to the instance connection addresses obtained in Prerequisites. Set hosts to the names of instance hosts. Specify a unique name for each host.

   For example:

   10.154.48.120 server01

   10.154.48.121 server02

   10.154.48.122 server03

3. Download client.truststore.jks. On the Kafka console, click the instance. Then on the instance details page, click Download next to SSL Certificate in the Connection area.

   Decompress the package to obtain the client certificate file client.truststore.jks.

4. Add the following commands in both the consumer.properties and producer.properties files (PLAIN is used as an example).

   sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
   username="**********" \
   password="**********";        
   sasl.mechanism=PLAIN
   
   security.protocol=SASL_SSL
   ssl.truststore.location={ssl_truststore_path}
   ssl.truststore.password=dms@kafka
   ssl.endpoint.identification.algorithm=

   Description:

   • username and password are specified when enabling SASL_SSL during instance creation.
   • ssl.truststore.location is the path for storing the certificate obtained in 3.
   • ssl.truststore.password is certified by the server, which must be set to dms@kafka and cannot be changed.
   • ssl.endpoint.identification.algorithm decides whether to verify the certificate domain name. This parameter must be left blank, which indicates disabling domain name verification.

Creating Messages

Go to the /bin directory of the Kafka client file and run the following command:

./kafka-console-producer.sh --broker-list ${connection addr} --topic ${topic name} --producer.config ../config/producer.properties

Description

• {connection-address}: the address obtained in Prerequisites.
• {topic-name}: the name of the topic created for the Kafka instance.

For example, 192.xxx.xxx.xxx:9093, 192.xxx.xxx.xxx:9093, 192.xxx.xxx.xxx:9093 are the connection addresses of the Kafka instance.

After running the preceding command, you can send a message to the Kafka instance by entering the information as prompted and pressing Enter. Contents in each line are sent as a message.

[root@ecs-kafka bin]#./kafka-console-producer.sh --broker-list 192.xxx.xxx.xxx:9093,192.xxx.xxx.xxx:9093,192.xxx.xxx.xxx:9093  --topic topic-demo --producer.config ../config/producer.properties
>Hello
>DMS
>Kafka!
>^C[root@ecs-kafka bin]# 

Press Ctrl+C to cancel.

Retrieving Messages

Run the following command:

./kafka-console-consumer.sh --bootstrap-server ${connection addr} --topic ${topic name} --group ${consumer group name} --from-beginning  --consumer.config ../config/consumer.properties

Description

• {connection-address}: the address obtained in Prerequisites.
• {topic-name}: the name of the topic created for the Kafka instance.
• {consumer-group-name}: the consumer group name set based on your service requirements. If a consumer group name has been specified in the configuration file, ensure that you use the same name in the command line. Otherwise, consumption may fail. If a consumer group name starts with a special character, such as an underscore (_) or a number sign (#), the monitoring data cannot be displayed.

Sample:

[root@ecs-kafka bin]#  ./kafka-console-consumer.sh --bootstrap-server 192.xxx.xxx.xxx:9093,192.xxx.xxx.xxx:9093,192.xxx.xxx.xxx:9093 --topic topic-demo --group order-test --from-beginning --consumer.config ../config/consumer.properties
Hello
Kafka!
DMS
^CProcessed a total of 3 messages
[root@ecs-kafka bin]# 

Press Ctrl+C to cancel.

Follow-Up Procedure

You can configure alarm rules for monitoring metrics to receive notifications in a timely manner when instances, brokers, or topics are abnormal.

Step 5: Configure Alarm Rules