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

Activity

1. RehanProgrammer

   added

   help wantedExtra attention is neededExtra attention is needed

   on Mar 25, 2025
2. ajitpal commented on Apr 17, 2025

   

   ajitpal

   on Apr 17, 2025

   More actions

   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 !