Getting Started with Kafka to Produce and Consume Messages

A Kafka instance runs in a Virtual Private Cloud (VPC). Before creating a Kafka instance, ensure that a VPC is available.

After a Kafka instance is created, download and install the Kafka open-source client on your ECS before creating and retrieving messages.

You can select the specification and quantity and enable SASL when creating a Kafka instance. Enabling SASL secures data transmission with encryption.

If automatic topic creation is not enabled during Kafka instance creation, you need to manually create topics for creating and retrieving messages.

Before connecting to a Kafka instance with SASL enabled, download the certificate and configure the connection in the client configuration file.

Configure the VPC and subnet for Kafka instances as required. You can use the current account's existing VPC and subnet, or create new ones. For details about how to create a VPC and a subnet, see Creating a VPC. Note that the VPC must be in the same region as the Kafka instance.

Configure the security group for Kafka instances as required. You can use the current account's existing security groups, or create new ones. For details about how to create a security group, see Creating a Security Group.

To connect to Kafka instances, add the security group rules described in Table 1. Other rules can be added based on site requirements.

Table 1 Security group rules

Direction	Protocol	Port	Source address	Description
Inbound	TCP	9093	0.0.0.0/0	Access Kafka instance in the same VPC over private network (SSL enabled).

After a security group is created, it has a default inbound rule that allows communication among ECSs within the security group and a default outbound rule that allows all outbound traffic. If you access your Kafka instance using the private network within a VPC, you do not need to add the rules described in Table 1.

1. Log in to the console, click in the upper left corner, click Elastic Cloud Server under Computing, and then create an ECS.

   For details about how to create an ECS, see Creating an ECS. If you already have an available ECS, skip this step.

2. Log in to the ECS.
3. Install Java JDK or JRE and configure the environment variables JAVA_HOME and PATH.
   Add the following lines to the .bash_profile file in the home directory as an authorized user. In this command, /opt/java/jdk1.8.0_151 is the JDK installation path. Change it to the path where you install JDK or JRE.

   export JAVA_HOME=/opt/java/jdk1.8.0_151 
   export PATH=$JAVA_HOME/bin:$PATH

   Run the source .bash_profile command for the modification to take effect.

   

   Use Oracle JDK instead of ECS's default JDK (for example, OpenJDK), because ECS's default JDK may not be suitable. Obtain Oracle JDK 1.8.111 or later from Oracle's official website.

4. Download an open-source Kafka client.

   wget https://archive.apache.org/dist/kafka/2.7.2/kafka_2.12-2.7.2.tgz

5. Run the following command to decompress the package:

   tar -zxf kafka_2.12-2.7.2.tgz

Kafka Manager is an open-source tool for managing Kafka clusters. After a Kafka instance is created, you can go to the instance details page to obtain the address for logging in to Kafka Manager. In Kafka Manager, you can view the monitoring and broker information of your Kafka clusters.

It takes 3 to 15 minutes to create an instance. During this period, the instance status is Creating.

• If the instance is created successfully, its status changes to Running.
• If the instance is in the Creation failed state, delete it. Then create a new one. If the instance creation fails again, contact customer service.
  

  Instances that fail to be created do not occupy other resources.

1. In the navigation pane, click Basic Information.
2. In the Connection area, view the connection address.
   Figure 2 Kafka instance addresses (private network) for intra-VPC access
   

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 1. 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 certificate.

   On the Kafka console, click the Kafka instance to go to the Basic Information 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 2.c.
   • 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. Here, this parameter must be left blank, which indicates disabling domain name verification.

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 addr}: the address obtained in 1.
• {topic name}: the topic name obtained in 4.

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.

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 addr}: the address obtained in 1.
• {topic name}: the topic name obtained in 4.
• {consumer group name}: the consumer group name set as required. 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.