title: Consumer consuming different topic data
body: <p><code>Context:</code>
  In our NodeJS app, we have setup 2 consumers having their respective topics eg: A1 &amp; A2. These consumers start when the service is initialized. we have added the initialization method call in index.js.</p>
  <p><code>Issue:</code> I have noticed that A1 consumer is consuming data from A2 topic, Ideally it should consume only the data present in A1 topic.</p>
  <p>I tried investing related to this but can't find this specific scenario.</p>
  <p><code>index.js</code></p>
  <pre><code>app.listen(app.get('port'), () =&gt; {
      logger.info(`Server started listening on port ${app.get('port')}.`);
      startReadingFromA1();
      startReadingFromA2();
  });
  
  initializeConsumer = async (
      logger: LogWrapper,
      clientId: string,
      topicName: string,
      callback: any,
      maxBytes: number = DEFAULT_KAFKA_CONSUMER_MAX_BYTES
  ) =&gt; {
      const country = topicName.split('-')[0];
      logger.info(`Initializing Consumer for clientId: ${clientId} and topicName: ${topicName}`);
      this.setClientId(clientId);
      let brokerList = this.getBrokerList();
      this.setBrokers(brokerList);
      this.setSasl(false);
      this.setSSL();
      this.setLogLevel(logLevel.ERROR);
      const kafkaConsumerObj = this.initialize();
      //group-&lt;country&gt;-&lt;module&gt;-&lt;env&gt;
      const groupId = `group-${topicName.toLowerCase()}-${process.env.NODE_ENV}`;
      if (!this.isConnected) {
          await this.connectToKafkaCluster(kafkaConsumerObj, groupId, maxBytes);
      }
      await this.subscribeConsumer(topicName);
      logger.info(`Connected Consumer Kafka cluster`);
      this.getMessage(logger, callback, country);
  }
  
  
  initialize = () =&gt; {
      return new Kafka({
          clientId: this.clientId,
          brokers: this.brokerList,
          sasl: this.sasl,
          ssl: this.ssl,
          logLevel: this.logLevel.ERROR,
          connectionTimeout: DEFAULT_CONNECTION_TIMEOUT,
          retry: {
              initialRetryTime: 100,
              retries: 3,
          },
      });
  };
  
  const startReadingFromA1 = async () =&gt; {
  const logger = LogHelper.getLoggerWithoutReq();
  try {
      logger.info(`Initializing the Kafka read from Topic`);
      const Consumer = KafkaConsumer.getInstance();
      const consumerId = nanoid();
      let countryInCaps = process.env.COUNTRY ? process.env.COUNTRY.toUpperCase() : null;
      await Consumer.initializeConsumer(logger, `CONS-${consumerId}`, `${countryInCaps}-ModuleA`, handleMessages);
      return true;
  } catch (err) {
      logger.error(`Error in startReadingFromA1: ${JSON.stringify(error)}`);
  }
  }
  
  
  const startReadingFromA2 = async () =&gt; {
  const logger = LogHelper.getLoggerWithoutReq();
  try {
      logger.info(`Initializing the Kafka read from Topic`);
      const Consumer = KafkaConsumer.getInstance();
      const consumerId = nanoid();
      let countryInCaps = process.env.COUNTRY ? process.env.COUNTRY.toUpperCase() : null;
      await Consumer.initializeConsumer(logger, `CONS-${consumerId}`, `${countryInCaps}-ModuleB`, handleFailedMessages);
      return true;
  } catch (err) {
      logger.error(`Error in startReadingFromA2: ${JSON.stringify(error)}`);
  }
  }
  </code></pre>
  <p>Thanks in advance.</p>