Fix issues with header conversions

Author: milindl

e2e/both.spec.js

@@ -424,12 +424,30 @@ describe('Consumer/Producer', function() {
     var headers = [
       { key1: 'value1' },
       { key2: Buffer.from('value2') },
-      { key3: 100 },
-      { key4: 10.1 },
     ];
     run_headers_test(done, headers);
   });
 
+  it('should not be able to produce any non-string and non-buffer headers: consumeLoop', function(done) {
+    producer.setPollInterval(10);
+
+    const headerCases = [
+      [ { key: 10 } ],
+      [ { key: null }],
+      [ { key: undefined }],
+    ];
+    for (const headerCase of headerCases) {
+      const buffer = Buffer.from('value');
+      const key = 'key';
+      t.throws(
+        () =>  producer.produce(topic, null, buffer, key, null, "", headerCase),
+        'must be string or buffer'
+      );
+    }
+
+    done();
+  });
+
   it('should be able to produce and consume messages: empty buffer key and empty value', function(done) {
     var emptyString = '';
     var key = Buffer.from(emptyString);

examples/consumer.js

@@ -75,6 +75,7 @@ async function consumerStart() {
       console.log({
         topic,
         partition,
+        headers: message.headers,
         offset: message.offset,
         key: message.key?.toString(),
         value: message.value.toString(),

examples/producer.js

@@ -14,8 +14,19 @@ async function producerStart() {
         res.push(producer.send({
             topic: 'test-topic',
             messages: [
-                { value: 'v222', partition: 1 },
-                { value: 'v11', partition: 0, key: 'x' },
+                {
+                    value: 'v1',
+                    partition: 0,
+                    key: 'x',
+                    headers: {
+                        'header1': ['h1v1', 'h1v2'],
+                        'header3': 'h3v3',
+                    }
+                },
+                {
+                    value: 'v2',
+                    key: 'y',
+                }
             ]
         }));
     }

examples/typescript/kafkajs.ts

@@ -22,7 +22,14 @@ async function runProducer() {
     await producer.send({
         topic: 'test-topic',
         messages: [
-            { value: 'Hello World!', key: 'key1' },
+            {
+                value: 'Hello World!',
+                key: 'key1',
+                headers: {
+                    'header1': 'value1',
+                    'header2': [Buffer.from('value2'), 'value3']
+                }
+            },
         ],
     });
 
@@ -52,6 +59,7 @@ async function runConsumer() {
             console.log({
                 key: message.key ? message.key.toString() : null,
                 value: message.value ? message.value.toString() : null,
+                headers: message.headers,
             });
         },
     });

lib/kafkajs/_common.js

@@ -559,12 +559,16 @@ function convertToRdKafkaHeaders(kafkaJSHeaders) {
 
   const headers = [];
   for (const [key, value] of Object.entries(kafkaJSHeaders)) {
-    if (value.constructor === Array) {
+    if (value && value.constructor === Array) {
       for (const v of value) {
-        headers.push({ key, value: v });
+        const header = {};
+        header[key] = v;
+        headers.push(header);
       }
     } else {
-      headers.push({ key, value });
+      const header = {};
+      header[key] = value;
+      headers.push(header);
     }
   }
   return headers;

lib/kafkajs/_consumer.js

@@ -608,13 +608,15 @@ class Consumer {
     let headers;
     if (message.headers) {
       headers = {}
-      for (const [key, value] of Object.entries(message.headers)) {
-        if (!Object.hasOwn(headers, key)) {
-          headers[key] = value;
-        } else if (headers[key].constructor === Array) {
-          headers[key].push(value);
-        } else {
-          headers[key] = [headers[key], value];
+      for (const header of message.headers) {
+        for (const [key, value] of Object.entries(header)) {
+          if (!Object.hasOwn(headers, key)) {
+            headers[key] = value;
+          } else if (headers[key].constructor === Array) {
+            headers[key].push(value);
+          } else {
+            headers[key] = [headers[key], value];
+          }
         }
       }
     }

src/producer.cc

@@ -595,18 +595,37 @@ NAN_METHOD(Producer::NodeProduce) {
 
         v8::Local<v8::Array> props = header->GetOwnPropertyNames(
           Nan::GetCurrentContext()).ToLocalChecked();
-        Nan::MaybeLocal<v8::String> v8Key = Nan::To<v8::String>(
-            Nan::Get(props, 0).ToLocalChecked());
-        Nan::MaybeLocal<v8::String> v8Value = Nan::To<v8::String>(
-            Nan::Get(header, v8Key.ToLocalChecked()).ToLocalChecked());
 
+        // TODO: Other properties in the list of properties should not be
+        // ignored, but they are. This is a bug, need to handle it either in JS
+        // or here.
+        Nan::MaybeLocal<v8::String> v8Key =
+            Nan::To<v8::String>(Nan::Get(props, 0).ToLocalChecked());
+
+        // The key must be a string.
+        if (v8Key.IsEmpty()) {
+          Nan::ThrowError("Header key must be a string");
+        }
         Nan::Utf8String uKey(v8Key.ToLocalChecked());
         std::string key(*uKey);
 
-        Nan::Utf8String uValue(v8Value.ToLocalChecked());
-        std::string value(*uValue);
-        headers.push_back(
-          RdKafka::Headers::Header(key, value.c_str(), value.size()));
+        // Valid types for the header are string or buffer.
+        // Other types will throw an error.
+        v8::Local<v8::Value> v8Value =
+            Nan::Get(header, v8Key.ToLocalChecked()).ToLocalChecked();
+
+        if (node::Buffer::HasInstance(v8Value)) {
+          const char* value = node::Buffer::Data(v8Value);
+          const size_t value_len = node::Buffer::Length(v8Value);
+          headers.push_back(RdKafka::Headers::Header(key, value, value_len));
+        } else if (v8Value->IsString()) {
+          Nan::Utf8String uValue(v8Value);
+          std::string value(*uValue);
+          headers.push_back(
+              RdKafka::Headers::Header(key, value.c_str(), value.size()));
+        } else {
+          Nan::ThrowError("Header value must be a string or buffer");
+        }
       }
     }
   }

test/promisified/consumer/consumeMessages.spec.js

@@ -85,6 +85,46 @@ describe.each([[true], [false]])('Consumer', (isAutoCommit) => {
         expect(messagesConsumed.map(m => m.message.offset)).toEqual(messages.map((_, i) => `${i}`))
     });
 
+    it('consume messages with headers', async () => {
+        await consumer.connect();
+        await producer.connect();
+        await consumer.subscribe({ topic: topicName })
+
+        const messagesConsumed = [];
+        consumer.run({ eachMessage: async event => messagesConsumed.push(event) });
+
+        const messages = [ {
+                value: `value-${secureRandom}`,
+                headers: {
+                    'header-1': 'value-1',
+                    'header-2': 'value-2',
+                    'header-3': [ 'value-3-1', 'value-3-2', Buffer.from([1,0,1,0,1]) ],
+                    'header-4': Buffer.from([1,0,1,0,1]),
+                }
+            } ]
+
+        await producer.send({ topic: topicName, messages })
+        await waitForMessages(messagesConsumed, { number: messages.length })
+
+        expect(messagesConsumed[0]).toEqual(
+            expect.objectContaining({
+                topic: topicName,
+                partition: 0,
+                message: expect.objectContaining({
+                    value: Buffer.from(messages[0].value),
+                    offset: '0',
+                    headers: {
+                        // Headers are always returned as Buffers from the broker.
+                        'header-1': Buffer.from('value-1'),
+                        'header-2': Buffer.from('value-2'),
+                        'header-3': [ Buffer.from('value-3-1'), Buffer.from('value-3-2'), Buffer.from([1,0,1,0,1]) ],
+                        'header-4': Buffer.from([1,0,1,0,1]),
+                    }
+                }),
+            })
+        )
+    });
+
     it.each([[true], [false]])('consumes messages using eachBatch', async (isAutoResolve) => {
         await consumer.connect();
         await producer.connect();