Merge pull request #114 from tulios/fix-produce-v3-decode

Fix producer v3 decode format
tulios · Aug 14, 2018 · a1e6c4b · a1e6c4b
2 parents 0f04b0b + cdef78e
commit a1e6c4b
Show file tree

Hide file tree

Showing 4 changed files with 100 additions and 11 deletions.
diff --git a/src/broker/__tests__/produce.spec.js b/src/broker/__tests__/produce.spec.js
@@ -158,7 +158,7 @@ describe('Broker > Produce', () => {
 
       const response1 = await broker2.produce({ topicData: createTopicData() })
       expect(response1).toEqual({
-        responses: [
+        topics: [
           {
             topicName,
             partitions: [{ baseOffset: '0', errorCode: 0, logAppendTime: '-1', partition: 0 }],
@@ -169,7 +169,7 @@ describe('Broker > Produce', () => {
 
       const response2 = await broker2.produce({ topicData: createTopicData() })
       expect(response2).toEqual({
-        responses: [
+        topics: [
           {
             topicName,
             partitions: [{ baseOffset: '3', errorCode: 0, logAppendTime: '-1', partition: 0 }],
@@ -195,7 +195,7 @@ describe('Broker > Produce', () => {
 
       const response1 = await broker2.produce({ topicData: createTopicData(true) })
       expect(response1).toEqual({
-        responses: [
+        topics: [
           {
             topicName,
             partitions: [{ baseOffset: '0', errorCode: 0, logAppendTime: '-1', partition: 0 }],
@@ -206,7 +206,7 @@ describe('Broker > Produce', () => {
 
       const response2 = await broker2.produce({ topicData: createTopicData() })
       expect(response2).toEqual({
-        responses: [
+        topics: [
           {
             topicName,
             partitions: [{ baseOffset: '3', errorCode: 0, logAppendTime: '-1', partition: 0 }],
@@ -236,7 +236,7 @@ describe('Broker > Produce', () => {
       })
 
       expect(response1).toEqual({
-        responses: [
+        topics: [
           {
             topicName,
             partitions: [{ baseOffset: '0', errorCode: 0, logAppendTime: '-1', partition: 0 }],
@@ -251,7 +251,7 @@ describe('Broker > Produce', () => {
       })
 
       expect(response2).toEqual({
-        responses: [
+        topics: [
           {
             topicName,
             partitions: [{ baseOffset: '3', errorCode: 0, logAppendTime: '-1', partition: 0 }],

diff --git a/src/producer/index.spec.js b/src/producer/index.spec.js
@@ -1,5 +1,4 @@
 const createProducer = require('./index')
-const { KafkaJSNonRetriableError } = require('../errors')
 const {
   secureRandom,
   connectionOpts,
@@ -11,6 +10,7 @@ const {
   sslBrokers,
   saslBrokers,
   newLogger,
+  testIfKafka011,
 } = require('testHelpers')
 
 const { KafkaJSSASLAuthenticationError } = require('../errors')
@@ -203,6 +203,95 @@ describe('Producer', () => {
     ])
   })
 
+  testIfKafka011('produce messages for Kafka 0.11', async () => {
+    const cluster = createCluster(
+      Object.assign(connectionOpts(), {
+        allowExperimentalV011: true,
+        createPartitioner: createModPartitioner,
+      })
+    )
+
+    producer = createProducer({ cluster, logger: newLogger() })
+    await producer.connect()
+
+    const sendMessages = async () =>
+      await producer.send({
+        acks: 1,
+        topic: topicName,
+        messages: new Array(10).fill().map((_, i) => ({
+          key: `key-${i}`,
+          value: `value-${i}`,
+        })),
+      })
+
+    expect(await sendMessages()).toEqual([
+      {
+        topicName,
+        baseOffset: '0',
+        errorCode: 0,
+        logAppendTime: '-1',
+        partition: 0,
+      },
+    ])
+
+    expect(await sendMessages()).toEqual([
+      {
+        topicName,
+        baseOffset: '10',
+        errorCode: 0,
+        logAppendTime: '-1',
+        partition: 0,
+      },
+    ])
+  })
+
+  testIfKafka011('produce messages for Kafka 0.11 with headers', async () => {
+    const cluster = createCluster(
+      Object.assign(connectionOpts(), {
+        allowExperimentalV011: true,
+        createPartitioner: createModPartitioner,
+      })
+    )
+
+    producer = createProducer({ cluster, logger: newLogger() })
+    await producer.connect()
+
+    const sendMessages = async () =>
+      await producer.send({
+        acks: 1,
+        topic: topicName,
+        messages: new Array(10).fill().map((_, i) => ({
+          key: `key-${i}`,
+          value: `value-${i}`,
+          headers: {
+            [`header-a${i}`]: `header-value-a${i}`,
+            [`header-b${i}`]: `header-value-b${i}`,
+            [`header-c${i}`]: `header-value-c${i}`,
+          },
+        })),
+      })
+
+    expect(await sendMessages()).toEqual([
+      {
+        topicName,
+        baseOffset: '0',
+        errorCode: 0,
+        logAppendTime: '-1',
+        partition: 0,
+      },
+    ])
+
+    expect(await sendMessages()).toEqual([
+      {
+        topicName,
+        baseOffset: '10',
+        errorCode: 0,
+        logAppendTime: '-1',
+        partition: 0,
+      },
+    ])
+  })
+
   test('produce messages to multiple topics', async () => {
     const topics = [`test-topic-${secureRandom()}`, `test-topic-${secureRandom()}`]
     const cluster = createCluster({

diff --git a/src/protocol/requests/produce/v3/response.js b/src/protocol/requests/produce/v3/response.js
@@ -23,21 +23,21 @@ const partition = decoder => ({
 
 const decode = async rawData => {
   const decoder = new Decoder(rawData)
-  const responses = decoder.readArray(decoder => ({
+  const topics = decoder.readArray(decoder => ({
     topicName: decoder.readString(),
     partitions: decoder.readArray(partition),
   }))
 
   const throttleTime = decoder.readInt32()
 
   return {
-    responses,
+    topics,
     throttleTime,
   }
 }
 
 const parse = async data => {
-  const partitionsWithError = data.responses.map(response => {
+  const partitionsWithError = data.topics.map(response => {
     return response.partitions.filter(partition => failure(partition.errorCode))
   })
 

diff --git a/src/protocol/requests/produce/v3/response.spec.js b/src/protocol/requests/produce/v3/response.spec.js
@@ -4,7 +4,7 @@ describe('Protocol > Requests > Produce > v3', () => {
   test('response', async () => {
     const data = await decode(Buffer.from(require('../fixtures/v3_response.json')))
     expect(data).toEqual({
-      responses: [
+      topics: [
         {
           partitions: [{ baseOffset: '0', errorCode: 0, logAppendTime: '-1', partition: 0 }],
           topicName: 'test-topic-ebba68879c6f5081d8c2',