Keyed kafka messages always seem to go to the same partition

Question

My node application uses the kafka-node node module.

I have a kafka topic with three partitions as seen below:

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

When I write a series of keyed messages to my topic, they all appear to be written to the same partition. I would expect some of my different keyed messages to be sent to partitions 1 and 2.

Here is my log output from the consumer onMessage event function for several messages:

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"}

Here is the kafka-node producer code to send a message:

  * @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');
    }
  }
}

As you can see, none of them are ever from partitions 1 & 2. This is true even after constantly sending messages with random integer keys for several minutes. What could I be doing wrong?

Answer 1

The correct partitionerType needs to be configured when you create the producer:

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

See the docs: https://www.npmjs.com/package/kafka-node#producerkafkaclient-options-custompartitioner