tencent cloud

cd /etc/yum.repos.d/

[Confluent.dist]
name=Confluent repository (dist)
baseurl=https://packages.confluent.io/rpm/5.1/7
gpgcheck=1
gpgkey=https://packages.confluent.io/rpm/5.1/archive.key
enabled=1   
[Confluent]
name=Confluent repository
baseurl=https://packages.confluent.io/rpm/5.1
gpgcheck=1
gpgkey=https://packages.confluent.io/rpm/5.1/archive.key
enabled=1

yum install librdkafka-devel

export CPPFLAGS=-I/usr/local/opt/openssl/include

export LDFLAGS=-L/usr/local/opt/openssl/lib

npm install i --unsafe-perm node-rdkafka

const Kafka = require('node-rdkafka');
const config = require('./setting');
console.log("features:" + Kafka.features);
console.log(Kafka.librdkafkaVersion);

var producer = new Kafka.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")

});

function produce() {
try {
   producer.produce(
   config['topic_name'],
   new Buffer('Hello CKafka SASL'),
   null,
   Date.now()
   );
} catch (err) {
   console.error('Error occurred when sending message(s)');
   console.error(err);
}
}

producer.on("disconnected", function() {
connected = false;
producer.connect();
})

producer.on('event.log', function(event) {
   console.log("event.log", event);
});

producer.on("error", function(error) {
   console.log("error:" + error);
});

producer.on('delivery-report', function(err, report) {
   console.log("delivery-report: producer ok");
});
// Any errors we encounter, including connection errors
producer.on('event.error', function(err) {
   console.error('event.error:' + err);
})

setInterval(produce,1000,"Interval");

node producer.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);
 });

node consumer.js