Fastify-kafkajs: support multiple different topics from different brokers.

[RehanProgrammer]

💬 Question here

Is it possible to add different brokers with fastyfy-kafkajs. When I try to add different brokers, it gives me the error "Not authorised to access the topic". I have read this #600, but I want to connect to 2 different brokers as we have 2 topics from different brokers

[ajitpal]

Please try below approaches if they help you

1. Create Separate Plugins for Each Broker:

   • Following the principle from Issue fastify-kafka: having multiple consumers or producers #600, you can define separate instances of fastify-kafka for each broker.

2. Example Implementation:

   const Fastify = require('fastify');
   const fastify = Fastify();
   const { Kafka } = require('kafkajs');
   
   // Plugin for Broker 1
   async function broker1Plugin(instance, opts) {
     const kafkaBroker1 = new Kafka({
       clientId: 'client1',
       brokers: ['broker1:9092'], // Replace with Broker 1 address
     });
   
     const consumer1 = kafkaBroker1.consumer({ groupId: 'group1' });
     await consumer1.connect();
     await consumer1.subscribe({ topic: 'topic1', fromBeginning: true });
   
     instance.decorate('consumer1', consumer1);
   
     await consumer1.run({
       eachMessage: async ({ topic, partition, message }) => {
         console.log(`Broker 1 - Topic: ${topic}, Message: ${message.value.toString()}`);
       },
     });
   }
   
   // Plugin for Broker 2
   async function broker2Plugin(instance, opts) {
     const kafkaBroker2 = new Kafka({
       clientId: 'client2',
       brokers: ['broker2:9092'], // Replace with Broker 2 address
     });
   
     const consumer2 = kafkaBroker2.consumer({ groupId: 'group2' });
     await consumer2.connect();
     await consumer2.subscribe({ topic: 'topic2', fromBeginning: true });
   
     instance.decorate('consumer2', consumer2);
   
     await consumer2.run({
       eachMessage: async ({ topic, partition, message }) => {
         console.log(`Broker 2 - Topic: ${topic}, Message: ${message.value.toString()}`);
       },
     });
   }
   
   // Register plugins
   fastify.register(broker1Plugin);
   fastify.register(broker2Plugin);
   
   fastify.listen(3000, (err, address) => {
     if (err) throw err;
     console.log(`Server listening at ${address}`);
   });

3. Notes:

   • Each Kafka instance is encapsulated within its respective plugin, ensuring isolation between brokers.
   • The example is inspired by the community-provided guidance in Issue fastify-kafka: having multiple consumers or producers #600, where creating separate instances was suggested.

4. Error Handling:

   • Check Kafka ACLs (Access Control Lists) for proper authorization for each broker and topic.

Feel free to adapt this solution to specific requirements. Let me know if this resolves your problem @RehanProgrammer !