Username in the format of instance ID + # + username. The instance ID can be obtained in Basic Info on the instance details page in the CKafka console, and the username is set when the user is created in ACL Policy Management > User Management.
sasl_plain_password
User password, which is set when the user is created in ACL Policy Management > User Management on the instance details page in the CKafka console.
bootstrap_servers
SASL access point. You can obtain it in Basic Info >Access Mode on the instance details page in the CKafka console.
topic_name
Topic name, which can be created and obtained in Topic Management on the instance details page in the CKafka console.
group_id
Consumer group ID, which can be customized as needed.
Step 4. Send messages
1. Write a message production program named producer.js.
const Kafka =require('node-rdkafka');
const config =require('./setting');
console.log("features:"+ Kafka.features);
console.log(Kafka.librdkafkaVersion);
var producer =newKafka.Producer({
'api.version.request':'true',
'bootstrap.servers': config['bootstrap_servers'],
'dr_cb':true,
'dr_msg_cb':true,
'security.protocol':'SASL_PLAINTEXT',
'sasl.mechanisms':'PLAIN',
'sasl.username': config['sasl_plain_username'],
'sasl.password': config['sasl_plain_password']
});
var connected =false
producer.setPollInterval(100);
producer.connect();
producer.on('ready',function(){
connected =true
console.log("connect ok")
});
functionproduce(){
try{
producer.produce(
config['topic_name'],
newBuffer('Hello CKafka SASL'),
null,
Date.now()
);
}catch(err){
console.error('Error occurred when sending message(s)');
// Any errors we encounter, including connection errors
producer.on('event.error',function(err){
console.error('event.error:'+ err);
})
setInterval(produce,1000,"Interval");
2. Run the following command to send messages.
node producer.js
3. View the execution result.
4. On the Topic Management tab on the instance details page in the CKafka console, select the target topic and click More > Message Query to view the message just sent.
Step 5. Subscribe to messages
1. Create the message consumption program consumer.js.
consumer.on('event.log', function(event){
console.log("event.log", event);
});
consumer.on('error', function(error){
console.log("error:" + error);
});
consumer.on('event', function(event){
console.log("event:" + event);
});const Kafka = require('node-rdkafka');
const config = require('./setting');
console.log(Kafka.features);
console.log(Kafka.librdkafkaVersion);
console.log(config)
var consumer = new Kafka.KafkaConsumer({
'api.version.request':'true',
'bootstrap.servers': config['bootstrap_servers'],
'security.protocol':'SASL_PLAINTEXT',
'sasl.mechanisms':'PLAIN',
'message.max.bytes':32000,
'fetch.message.max.bytes':32000,
'max.partition.fetch.bytes':32000,
'sasl.username': config['sasl_plain_username'],
'sasl.password': config['sasl_plain_password'],
'group.id': config['group_id']
});
consumer.connect();
consumer.on('ready', function(){
console.log("connect ok");
consumer.subscribe([config['topic_name']]);
consumer.consume();
})
consumer.on('data', function(data){
console.log(data);
});
2. Execute the following command to send messages.
node consumer.js
3. View the execution result.
4. On the Consumer Group page in the Ckafka console, click the triangle icon on the left of the target consumer group name, enter the topic name in the search box, and click View Details to view the consumption details.
Yes
No
Was this page helpful?