'key' in message received via Consumer hasn't been decoded.

[twang2218]

Bug Report

Consumer.on('message'), I expect the message.key should be a string, just as same as what I send via Producer.send(), however, what I got is a Buffer

Environment

• Node version: v8.1.3
• Kafka-node version: 1.6.2
• Kafka version: v0.10.2.1

For specific cases also provide

• Number of Brokers: 1
• Number partitions for topic: 1

Include Sample Code to reproduce behavior

const kafka = require('kafka-node');

const { HighLevelConsumer, Client } = kafka;

const client = new Client('localhost:2181', 'my-client-id2');

client.on('error', error => console.error(error));

const consumer = new HighLevelConsumer(client, [{ topic: 'node-test7' }], {
  groupId: 'my-group-1',
  autoCommit: true,
  fromOffset: 0,
  encoding: 'utf8',
});

consumer.on('message', message => console.log(`> '${JSON.stringify(message)}'`));
consumer.on('error', error => console.error(error));

process.on('SIGING', () => consumer.close(true, () => process.exit()));

Include output with Debug turned on

  kafka-node:zookeeper Node: /consumers/my-group-1/ids/my-group-1_464fef47-804e-4329-8832-6c0de40f0c2e was created. +0ms
  kafka-node:HighLevelConsumer rebalance() my-group-1_464fef47-804e-4329-8832-6c0de40f0c2e is rebalancing: false ready: false +4ms
  kafka-node:HighLevelConsumer Deregistered listeners my-group-1_464fef47-804e-4329-8832-6c0de40f0c2e +2ms
  kafka-node:HighLevelConsumer Registered listeners my-group-1_464fef47-804e-4329-8832-6c0de40f0c2e +0ms
  kafka-node:HighLevelConsumer HighLevelConsumer rebalance retry config: {"retries":10,"factor":2,"minTimeout":100,"maxTimeout":1000,"randomize":true} +1ms
  kafka-node:HighLevelConsumer HighLevelConsumer my-group-1_464fef47-804e-4329-8832-6c0de40f0c2e is attempting to rebalance +25ms
  kafka-node:HighLevelConsumer HighLevelConsumer my-group-1_464fef47-804e-4329-8832-6c0de40f0c2e stopping data read during rebalance +0ms
  kafka-node:HighLevelConsumer HighLevelConsumer my-group-1_464fef47-804e-4329-8832-6c0de40f0c2e assembling data for rebalance +52ms
  kafka-node:zookeeper Children are: ["my-group-1_464fef47-804e-4329-8832-6c0de40f0c2e"]. +9ms
  kafka-node:HighLevelConsumer HighLevelConsumer my-group-1_464fef47-804e-4329-8832-6c0de40f0c2e releasing current partitions during rebalance +10ms
  kafka-node:HighLevelConsumer HighLevelConsumer my-group-1_464fef47-804e-4329-8832-6c0de40f0c2e determining the partitions to own during rebalance +2ms
  kafka-node:HighLevelConsumer consumerPerTopicMap.consumerTopicMap {"my-group-1_464fef47-804e-4329-8832-6c0de40f0c2e":["node-test7"]} +0ms
  kafka-node:HighLevelConsumer newTopicPayloads [{"topic":"node-test7","partition":"0","offset":0,"maxBytes":1048576,"metadata":"m"}] +0ms
  kafka-node:HighLevelConsumer HighLevelConsumer my-group-1_464fef47-804e-4329-8832-6c0de40f0c2e gaining ownership of partitions during rebalance +1ms
  kafka-node:zookeeper Gained ownership of /consumers/my-group-1/owners/node-test7/0 by my-group-1_464fef47-804e-4329-8832-6c0de40f0c2e. +107ms
  kafka-node:HighLevelConsumer HighLevelConsumer my-group-1_464fef47-804e-4329-8832-6c0de40f0c2e rebalance attempt was successful +1ms
  kafka-node:Client refresh metadata currentAttempt 1 +1ms
  kafka-node:HighLevelConsumer Registered listeners my-group-1_464fef47-804e-4329-8832-6c0de40f0c2e +5ms
  kafka-node:HighLevelConsumer in fetchOffset my-group-1_464fef47-804e-4329-8832-6c0de40f0c2e payloads: [{"topic":"node-test7","partition":"0","offset":0,"maxBytes":1048576,"metadata":"m"}] +0ms
  kafka-node:HighLevelConsumer Deregistered listeners my-group-1_464fef47-804e-4329-8832-6c0de40f0c2e +24ms
> '{"topic":"node-test7","value":"message 1","offset":14,"partition":0,"highWaterOffset":16,"key":{"type":"Buffer","data":[108,111,119,107,101,121]}}'
> '{"topic":"node-test7","value":"message 1","offset":15,"partition":0,"highWaterOffset":16,"key":{"type":"Buffer","data":[104,105,103,104,107,101,121]}}'
> '{"topic":"node-test7","value":"message 2","offset":16,"partition":0,"highWaterOffset":18,"key":{"type":"Buffer","data":[108,111,119,107,101,121]}}'
> '{"topic":"node-test7","value":"message 2","offset":17,"partition":0,"highWaterOffset":18,"key":{"type":"Buffer","data":[104,105,103,104,107,101,121]}}'
> '{"topic":"node-test7","value":"message 3","offset":18,"partition":0,"highWaterOffset":20,"key":{"type":"Buffer","data":[108,111,119,107,101,121]}}'
> '{"topic":"node-test7","value":"message 3","offset":19,"partition":0,"highWaterOffset":20,"key":{"type":"Buffer","data":[104,105,103,104,107,101,121]}}'
...

As you can see, the key are in form of {"type":"Buffer","data":[104,105,103,104,107,101,121]}, which I expect it to be a string like highkey.

[twang2218]

Is the parsing code here?

kafka-node/lib/protocol/protocol.js

Lines 156 to 161 in 7be0e1c

	.word32bs('key')
	.tap(function (vars) {
	if (vars.key === -1) return;
	cur += vars.key;
	this.buffer('key', vars.key);
	})

      .word32bs('key')
      .tap(function (vars) {
        if (vars.key === -1) return;
        cur += vars.key;
        this.buffer('key', vars.key);
      })

Should we parse it as a string instead?

[hyperlink]

We should respect the consumer's encoding option or create a new option something like keyEncoding and use it if provided. Since key has always returned an Buffer previously any changes made to this default behavior would become a breaking change.

[hyperlink]

Should be decoded as a string now by default. Fixed in #704 and published as 2.0.0.