我如何在令牌到期之前使用sasl与oauthbearer为kafka生产者和使用kafkajs的消费者定期获得新的acc

Question

const axios = require('axios');
const dotenv = require('dotenv');
dotenv.config();
const { Kafka } = require('kafkajs');
var qs = require('qs');
var data = qs.stringify({
'client_id': process.env.CLIENT_ID,
'client_secret': process.env.CLIENT_SECRET,
'grant_type': process.env.GRANT_TYPE 
});
var config = {
method: 'post',
url: process.env.OAUTH_TOKEN_PROVIDER_URL,
headers: { 
'Content-Type': 'application/x-www-form-urlencoded'
},
data : data
};
async function getToken(){
let token = await axios(config)
.then((response) => {
return response.data.access_token;
})
.catch((error) => {
console.log(error);
});
return token;
}
const kafka = new Kafka({
clientId: process.env.CLIENT_ID,
clientSecret:process.env.CLIENT_SECRET,
brokers: [process.env.BROKER_URL],
retry: {
initialRetryTime: 100,
retries: 2
},
// authenticationTimeout: 1000,
// reauthenticationThreshold: 10000,
ssl: true,
sasl: {
mechanism: 'oauthbearer',
oauthBearerProvider: async () => {
const token = await getToken() //this is the function to get token
return {
value: token
}
}
},
});
const topic = 'topic_test';
const producer = kafka.producer();
async function produceMesage(){
try {
await producer.send({
topic: topic,
messages: [
{ key: 'key1', value: message, partition: 0 }
],
})
} catch (error) {
console.log(error);
//I want to avoid this error when token expires
if(error.type =='TOPIC_AUTHORIZATION_FAILED') producer.connect();
} 
}
const run = async () => {
await producer.connect()
setInterval(produceMesage, 1000)
}
run().catch(e => console.error(`[example/producer] ${e.message}`, e))

这是我的代码来产生事件，它工作得很好。我有一个getToken()函数来获取access_token来验证生产者/消费者连接。问题是我使用getToken()函数获得的accesss_token的过期时间为32400秒或9小时，我想通过在令牌到期之前获得新令牌来保持生产者运行，最好的方法是什么?

Answer 1

希望这篇文章能对某个人有所帮助

const { Kafka } = require('kafkajs');
async function startProducer(kafka, accessToken) {
try {
// Create the producer
const producer = kafka.producer();
// Connect the producer to the Kafka cluster
await producer.connect();
// Set up the request options for the producer
const request = {
topic: 'my-topic',
messages: [{ key: 'key', value: 'value' }],
accessToken: { value: accessToken }  // pass the access token as a request header
};
// Send the message
await producer.send(request);
// Close the producer
await producer.disconnect();
} catch (error) {
console.error(error);
}
}
async function main() {
// Set up the Kafka client
const kafka = new Kafka({
clientId: 'my-client',
brokers: ['kafka1:9092', 'kafka2:9092']
});
// Get the initial access token
let accessToken = await getAccessToken();
// Start the producer with the initial access token
await startProducer(kafka, accessToken);
// Set up a timer to refresh the access token before it expires
setInterval(async () => {
// Check if the access token has expired
if (Date.now() >= accessToken.expirationTime) {
// Get a new access token
accessToken = await getAccessToken(refreshToken);
// Restart the producer with the new access token
await startProducer(kafka, accessToken);
}
}, 60000);  // check every 60 seconds
}
main().catch(console.error);