node.js - 3 つ以上のトピックをサブスクライブすると Kafka コンシューマーがメッセージを受信しない

Question

node-rdkafka ( https://github.com/Blizzard/node-rdkafka ) を使用して、IBM Bluemix Message Hub からメッセージを消費および生成しています。ノード プロセスごとに最大 2 つのトピックのみをサブスクライブする限り、問題なくメッセージを消費および生成できます。3 つ以上のトピックをサブスクライブするとすぐに、コンシューマーはサブスクライブされたトピックに関するメッセージを受信しなくなります。エラーはありません。

ここにソフトリミットはありますか？または、私のコードにこの問題の原因となっているものがありますか? サーバーのメモリを増やしても効果がないようです。

プロデューサー コード:

events.send = events.produce = (topic, type, data) => {
  log.info('Sending message on topic ' + topic);

  let producer = lib.getProducer(hubConfig);

  // Connect to the broker manually
  producer.connect({}, (err) => {
    if (err) {
      log.error('Producer failed to connect');
      log.error(err);
    }
  });

  // Wait for the ready event before proceeding
  producer.on('ready', () => {
    log.info('Producer ready, sending message');
    try {
      producer.produce(
        topic,
        null,
        new Buffer(JSON.stringify(data)),
        type,
        Date.now()
      );
    } catch (err) {
      log.error('A problem occurred when sending our message');
      log.error(err);
    }
  });

  producer.on('event.error', (err) => {
    log.error('Error from producer');
    log.error(err);
  })
};

lib.getProducer = (hubConfig) => {
  return new Kafka.Producer({
      'metadata.broker.list': hubConfig.kafka_brokers_sasl.join(','),
      'security.protocol': 'sasl_ssl',
      'ssl.ca.location': '/etc/ssl/certs',
      'sasl.mechanisms': 'PLAIN',
      'sasl.username': hubConfig.user,
      'sasl.password': hubConfig.password,
      'api.version.request': true,
      'dr_cb': true,
      'event_cb': true
    });
};

消費者：

events.listen = events.consume = (topics, callback) => {
    if (!_.isArray(topics)) {
      topics = [topics];
    }
    log.info('Subscribing to ' + topics.join(', ') + ' on test event listener...');
    let consumer,
      emitter = new evt.EventEmitter(),

      // Each consumer has a unique group and client ID
      groupName = 'group-' + uuidv1(),
      clientName = 'client-' + uuidv1();

    consumer = lib.getConsumer(hubConfig, groupName, clientName);

    consumer.connect({}, (err) => {
      if (err) {
        log.error('Consumer failed to connect');
        log.error(err);
        if (callback) callback(err);
      }
    });
    consumer
      .on('ready', function() {
        log.info('Consumer connected, subscribed to ' + topics.join(', '));
        consumer.subscribe(topics);
        consumer.consume();
        if (callback) callback();
      })
      .on('data', function(data) {
        let d = data.value.toString().replace(/"/g,''),
          dupeKey = d + '-' + data.key;
        if (!duplicateBuffer[dupeKey]) {
          emitter.emit('message', {
            data: d,
            type: data.key,
            topic: data.topic
          });

          duplicateBuffer[dupeKey] = setTimeout(() => {
            delete duplicateBuffer[dupeKey];
          }, DUPE_DELAY);
        } else {
          log.info('Ignoring duplicate event: ' + d + ' ' + data.type);
        }
      })
      .on('error', (err) => {
        log.error(err);
        emitter.emit('error', err);
      });

    return emitter;
  };

lib.getConsumer = (hubConfig, groupName, clientName) => {
  return new Kafka.KafkaConsumer({
      'group.id': groupName,
      'client.id': clientName,
      'metadata.broker.list': hubConfig.kafka_brokers_sasl.join(','),
      'security.protocol': 'sasl_ssl',
      'ssl.ca.location': '/etc/ssl/certs',
      'sasl.mechanisms': 'PLAIN',
      'sasl.username': hubConfig.user,
      'sasl.password': hubConfig.password,
      'api.version.request': true,
      'event_cb': true
    }, {});
};

助言がありますか？