Connecting to a Topic

Updated on 2024-11-27 GMT+08:00

View PDF

Overview

After a topic is created, you can use the CLI or open-source Kafka client to connect to the topic and produce and consume messages in the topic.

This section describes how to connect to a topic using CLI. If you use an open-source Kafka client, see Configuring MQS Connection (Open-Source Client).

Prerequisites

A topic is available. If no topic is available, create a topic first.
You have downloaded the open-source Kafka CLI tool whose version matches the Kafka version of the ROMA Connect instance.
You can view the Kafka version in the MQS Information area on the Instance Information page of the ROMA Connect console.
- Kafka command line tool 1.1.0
- Kafka command line tool 2.7.2
Java JDK has been installed in the environment where the Kafka command-line tool is used, and related environment variables have been configured.
If MQS SASL_SSL has been enabled for the ROMA Connect instance, ensure that the client certificate has been downloaded. To download the client certificate, log in to the ROMA Connect console, choose Message Queue Service > Topic Management, and click Download SSL Certificate.
If both SASL_SSL and intra-VPC plaintext access are enabled for MQS of the ROMA Connect instance, the SASL mode cannot be used for connecting to MQS topics in the VPC.
If the SASL mode is used for connecting to MQS topics, you are advised to configure the mapping between the host and IP address in the /etc/hosts file on the host where the client is located. Otherwise, network delay will occur.
Set the IP address to the connection address of MQS. Set the host to the name of each instance host. Ensure that the name of each host is unique. For example:

10.10.10.11 host01

10.10.10.12 host02

10.10.10.13 host03
A maximum of 500 consumers in the same consumer group can connect to the same MQS. If the number of consumers exceeds 500, the connection fails. If a consumer group with over 500 consumers needs to connect to an MQS, put the consumers into multiple consumer groups.

Using SASL Authentication

If SASL_SSL access is enabled for the ROMA Connect instance, messages produced and consumed by the client to the topic are encrypted for transmission, which is more secure. The following uses Linux as an example.

Decompress the Kafka command-line tool and client certificates.
Access the directory where the tool package is stored and run the following command to decompress the package:
- Run the following command to decompress the command-line tool package:
```
tar -zxf kafka_tar
```
  In the preceding command, kafka_tar indicates the name of the Kafka CLI package.
- Run the following command to decompress the client certificate file package:
```
unzip cert_zip
```
  In the preceding command, cert_zip indicates the name of the client certificate file package.
Modify the configuration file of the Kafka command-line tool.
Find the consumer.properties and producer.properties files in the /config directory of the Kafka command-line tool and add the following content to the files:
```
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
username="**********" \
password="**********";
sasl.mechanism=PLAIN
security.protocol=SASL_SSL
ssl.truststore.location=/cert/client.truststore.jks
ssl.truststore.password=dms@kafka
ssl.endpoint.identification.algorithm=
```
- The values of username and password are the key and secret of the integration application to which the topic belongs. For details on how to obtain the key and secret, see Viewing and Editing an Integration Application.
- The value of ssl.truststore.location is the path for storing the client certificate obtained after decompression in 1. Replace it with the actual path. Note that the certificate path in Windows must contain slashes (/).
- ssl.truststore.password indicates the server certificate password, which must be set to dms@kafka and cannot be changed.
Access the /bin directory of the Kafka command-line tool.
In the Windows operating system, you need to access the /bin/windows directory.
Produce a message to the topic.
1. Run the following command to create a connection with the topic for producing messages:
```
./kafka-console-producer.sh --broker-list Address --topic TopicName --producer.config ../config/producer.properties
```
  In the preceding command:
  - Address indicates the MQS address of ROMA Connect. For details about how to obtain the MQS address, see Viewing Details of an Instance. If you access the topic from a public network, use the public connection address. If you access the topic using the intra-VPC mode, use the intranet connection address.
  - TopicName indicates the name of the topic where the message is to be produced.
  - ../config/producer.properties indicates the relative path of the configuration file.
2. Enter the message contents and send the messages to the topic.
```
>Message1
>Message2
>Message3
```
  In the preceding information, Message1, Message2, and Message3 indicate the actual message contents sent to the topic. Each row indicates a message.
3. To disconnect from the topic, press Ctrl+C.
Consume messages from a topic.
NOTE:

When a consumer consumes messages from multiple partitions of a topic, messages in only one partition can be consumed at a time. Messages in multiple partitions are consumed in multiple times.
1. Run the following command to create a connection with the topic for reading messages:
```
./kafka-console-consumer.sh --bootstrap-server Address --topic TopicName --from-beginning --consumer.config ../config/consumer.properties
```
  In the preceding command:
  - Address indicates the MQS address of ROMA Connect. For details about how to obtain the MQS address, see Viewing Details of an Instance. If you access the topic from a public network, use the public connection address. If you access the topic using the intra-VPC mode, use the intranet connection address.
  - TopicName indicates the name of the topic where the message is to be consumed.
  - ../config/consumer.properties indicates the relative path of the configuration file.
2. After the command is executed, the system is continuously connected to the topic and reads messages. To disconnect from the topic, press Ctrl+C.

Without Using SASL Authentication

If SASL_SSL access is not enabled for the ROMA Connect instance, the client does not need to load the certificate. Messages produced and consumed by the topic are not encrypted. The following uses Linux as an example.

Decompress the Kafka command-line tool.
Access the directory where the tool package is stored and run the following command to decompress the package:
```
tar -zxf kafka_tar
```
In the preceding command, kafka_tar indicates the name of the Kafka CLI package.
Access the /bin directory of the Kafka command-line tool.
In the Windows operating system, you need to access the /bin/windows directory.
Produce a message to the topic.
1. Run the following command to create a connection with the topic for producing messages:
```
./kafka-console-producer.sh --broker-list Address --topic TopicName
```
  In the preceding command:
  - Address indicates the MQS address of ROMA Connect. For details about how to obtain the MQS address, see Viewing Details of an Instance. If you access the topic from a public network, use the public connection address. If you access the topic using the intra-VPC mode, use the intranet connection address.
  - TopicName indicates the name of the topic where the message is to be produced.
2. Enter the message contents and send the messages to the topic.
```
>Message1
>Message2
>Message3
```
  In the preceding information, Message1, Message2, and Message3 indicate the actual message contents sent to the topic. Each row indicates a message.
3. To disconnect from the topic, press Ctrl+C.
Consume messages from a topic.
NOTE:

When a consumer consumes messages from multiple partitions of a topic, messages in only one partition can be consumed at a time. Messages in multiple partitions are consumed in multiple times.
1. Run the following command to create a connection with the topic for reading messages:
```
./kafka-console-consumer.sh --bootstrap-server Address --topic TopicName --from-beginning
```
  In the preceding command:
  - Address indicates the MQS address of ROMA Connect. For details about how to obtain the MQS address, see Viewing Details of an Instance. If you access the topic from a public network, use the public connection address. If you access the topic using the intra-VPC mode, use the intranet connection address.
  - TopicName indicates the name of the topic where the message is to be produced.
2. After the command is executed, the system is continuously connected to the topic and reads messages. To disconnect from the topic, press Ctrl+C.