Сообщения с ключом kafka всегда кажутся go тому же разделу - PullRequest
Новая
       31

Сообщения с ключом kafka всегда кажутся go тому же разделу

1 голос
/ 19 марта 2020

Приложение моего узла использует модуль узла kafka-node.

У меня есть kafka topi c с тремя разделами, как показано ниже: 

Topic: NotifierTemporarye3df:/opPartitionCount: 3in$ kafReplicationFactor: 3ibe Configs: segment.bytes=1073741824 --topic NotifierTemporary
    Topic: NotifierTemporary        Partition: 0    Leader: 1001    Replicas: 1001,1003,1002        Isr: 1001,1003,1002
    Topic: NotifierTemporary        Partition: 1    Leader: 1002    Replicas: 1002,1001,1003        Isr: 1002,1001,1003
    Topic: NotifierTemporary        Partition: 2    Leader: 1003    Replicas: 1003,1002,1001        Isr: 1003,1002,1001

Когда я пишу серию сообщения на мой topi c, все они, кажется, записаны в один и тот же раздел. Я ожидаю, что некоторые из моих различных сообщений с ключами будут отправлены в разделы 1 и 2.

Вот мой вывод журнала из функции события onMessage потребителя для нескольких сообщений: 

the message is: {"topic":"NotifierTemporary","value":"{\"recipient\":66,\"subject\":\"download complete\",\"message\":\"s3/123.jpg\"}","offset":345,"partition":0,"highWaterOffset":346,"key":"66","timestamp":"2020-03-19T00:16:57.783Z"}
the message is: {"topic":"NotifierTemporary","value":"{\"recipient\":222,\"subject\":\"download complete\",\"message\":\"s3/123.jpg\"}","offset":346,"partition":0,"highWaterOffset":347,"key":"222","timestamp":"2020-03-19T00:16:57.786Z"}
the message is: {"topic":"NotifierTemporary","value":"{\"recipient\":13,\"subject\":\"download complete\",\"message\":\"s3/123.jpg\"}","offset":347,"partition":0,"highWaterOffset":348,"key":"13","timestamp":"2020-03-19T00:16:57.791Z"}
the message is: {"topic":"NotifierTemporary","value":"{\"recipient\":316,\"subject\":\"download complete\",\"message\":\"s3/123.jpg\"}","offset":348,"partition":0,"highWaterOffset":349,"key":"316","timestamp":"2020-03-19T00:16:57.798Z"}
the message is: {"topic":"NotifierTemporary","value":"{\"recipient\":446,\"subject\":\"download complete\",\"message\":\"s3/123.jpg\"}","offset":349,"partition":0,"highWaterOffset":350,"key":"446","timestamp":"2020-03-19T00:16:57.806Z"}
the message is: {"topic":"NotifierTemporary","value":"{\"recipient\":66,\"subject\":\"download complete\",\"message\":\"s3/123.jpg\"}","offset":350,"partition":0,"highWaterOffset":351,"key":"66","timestamp":"2020-03-19T00:17:27.918Z"}
the message is: {"topic":"NotifierTemporary","value":"{\"recipient\":222,\"subject\":\"download complete\",\"message\":\"s3/123.jpg\"}","offset":351,"partition":0,"highWaterOffset":352,"key":"222","timestamp":"2020-03-19T00:17:27.920Z"}
the message is: {"topic":"NotifierTemporary","value":"{\"recipient\":13,\"subject\":\"download complete\",\"message\":\"s3/123.jpg\"}","offset":352,"partition":0,"highWaterOffset":353,"key":"13","timestamp":"2020-03-19T00:17:27.929Z"}
the message is: {"topic":"NotifierTemporary","value":"{\"recipient\":316,\"subject\":\"download complete\",\"message\":\"s3/123.jpg\"}","offset":353,"partition":0,"highWaterOffset":354,"key":"316","timestamp":"2020-03-19T00:17:27.936Z"}

Вот код производителя узла kafka для отправки сообщения: 

  * @description Adds a notification message to the Kafka topic that is not saved in a database.
  * @param {Int} recipientId - accountId of recipient of notification message
  * @param {Object} message - message payload to send to recipient
  */
  async sendTemporaryNotification(recipientId, subject, message) {
    const notificationMessage = {
      recipient: recipientId,
      subject,
      message,
    };
    // we need to validate this message schema - this will throw if invalid
    Joi.assert(notificationMessage, NotificationMessage);
    // partition based on the recipient
    const payloads = [
      { topic: KAFKA_TOPIC_TEMPORARY, messages: JSON.stringify(notificationMessage), key: notificationMessage.recipient },
    ];
    if (this.isReady) {
      await this.producer.sendAsync(payloads);
    }
    else {
      throw new ProducerNotReadyError('Notifier Producer not ready');
    }
  }
}

Как видите, ни один из них никогда не был из разделов 1 и 2. Это верно даже после постоянной отправки сообщений со случайными целочисленными ключами для нескольких минут. Что я могу делать не так?

Ответы [ 2 ]

2 голосов
/ 21 марта 2020

Правильное partitionerType необходимо настроить при создании производителя: 

// Partitioner type (default = 0, random = 1, cyclic = 2, keyed = 3, custom = 4), default 0
new Producer(client, {paritionerType: 3});

См. Документы: https://www.npmjs.com/package/kafka-node#producerkafkaclient -options-custompartitioner

1 голос
/ 22 марта 2020

Scarysize был прав насчет того, что я не указал тип разделителя. Для тех, кто интересуется, как выглядит полный разделенный производитель, вы можете обратиться к этому коду. Я проверил, что это распространяет сообщения на основе предоставленных ключей. Я использовал HighLevelProducer здесь, потому что один из основных разработчиков библиотеки kafka-node предложил другим использовать ее для решения проблем секционирования. Я не проверял, что это решение будет работать с обычным источником, а не с HighLevelProducer.

В этом примере я отправляю уведомления пользователям на основе их идентификатора пользователя. Это ключ, на который делятся сообщения. 

const { KafkaClient, HighLevelProducer, KeyedMessage } = require('kafka-node');
const Promise = require('bluebird');
const NotificationMessage = require(__dirname + '/../models/notificationMessage.js');
const ProducerNotReadyError = require(__dirname + '/../errors/producerNotReadyError.js');
const Joi = require('@hapi/joi');

const KAFKA_TOPIC_TEMPORARY = 'NotifierTemporary';

/**
 * @classdesc Producer that sends notification messages to Kafka.
 * @class
 */
class NotifierProducer {

  /**
  * Create NotifierProducer.
  * @constructor
  * @param {String} kafkaHost - address of kafka server
  */
  constructor(kafkaHost) {
    const client = Promise.promisifyAll(new KafkaClient({kafkaHost}));
    const producerOptions = {
      partitionerType: HighLevelProducer.PARTITIONER_TYPES.keyed, // this is a keyed partitioner
    };
    this.producer = Promise.promisifyAll(new HighLevelProducer(client, producerOptions));
    this.isReady = false;

    this.producer.on('ready', async () => {
      await client.refreshMetadataAsync([KAFKA_TOPIC_TEMPORARY]);
      console.log('Notifier Producer is operational');
      this.isReady = true;
    });

    this.producer.on('error', err => {
      console.error('Notifier Producer error: ', err);
      this.isReady = false;
    });
  }
  /**
  * @description Adds a notification message to the Kafka topic that is not saved in a database.
  * @param {Int} recipientId - accountId of recipient of notification message
  * @param {String} subject - subject header of the message
  * @param {Object} message - message payload to send to recipient
  */
  async sendTemporaryNotification(recipientId, subject, message) {
    const notificationMessage = {
      recipient: recipientId,
      subject,
      message,
    };
    // we need to validate this message schema - this will throw if invalid
    Joi.assert(notificationMessage, NotificationMessage);
    // partition based on the recipient
    const messageKM = new KeyedMessage(notificationMessage.recipient, JSON.stringify(notificationMessage));
    const payloads = [
      { topic: KAFKA_TOPIC_TEMPORARY, messages: messageKM, key: notificationMessage.recipient },
    ];
    if (this.isReady) {
      await this.producer.sendAsync(payloads);
    }
    else {
      throw new ProducerNotReadyError('Notifier Producer not ready');
    }
  }
}

/**
 * Kafka topic that the producer and corresponding consumer will use to send temporary messages.
 * @type {string}
*/
NotifierProducer.KAFKA_TOPIC_TEMPORARY = KAFKA_TOPIC_TEMPORARY;

module.exports = NotifierProducer;
...