Prevent inflight's correlation id collisions

src/errors.js

@@ -178,6 +178,14 @@ class KafkaJSUnsupportedMagicByteInMessageSet extends KafkaJSNonRetriableError {
   }
 }
 
+class KafkaJSCorrelationIdAlreadyExists extends KafkaJSNonRetriableError {
+  constructor(correlationId) {
+    super(...arguments)
+    this.name = 'KafkaJSCorrelationIdAlreadyExists'
+    this.message = `Correlation id ${correlationId} already exists`
+  }
+}
+
 module.exports = {
   KafkaJSError,
   KafkaJSNonRetriableError,
@@ -201,4 +209,5 @@ module.exports = {
   KafkaJSLockTimeout,
   KafkaJSServerDoesNotSupportApiKey,
   KafkaJSUnsupportedMagicByteInMessageSet,
+  KafkaJSCorrelationIdAlreadyExists,
 }

src/network/requestQueue/index.js

@@ -1,6 +1,7 @@
 const EventEmitter = require('events')
 const SocketRequest = require('./socketRequest')
 const events = require('../instrumentationEvents')
+const { KafkaJSCorrelationIdAlreadyExists } = require('../../errors')
 
 const PRIVATE = {
   EMIT_QUEUE_SIZE_EVENT: Symbol('private:RequestQueue:emitQueueSizeEvent'),
@@ -131,6 +132,9 @@ module.exports = class RequestQueue extends EventEmitter {
       instrumentationEmitter: this.instrumentationEmitter,
       requestTimeout,
       send: () => {
+        if (this.inflight.has(correlationId)) {
+          throw new KafkaJSCorrelationIdAlreadyExists(correlationId)
+        }
         this.inflight.set(correlationId, socketRequest)
         pushedRequest.sendRequest()
       },

src/network/requestQueue/index.spec.js

@@ -5,6 +5,7 @@ const { newLogger } = require('testHelpers')
 const InstrumentationEventEmitter = require('../../instrumentation/emitter')
 const events = require('../instrumentationEvents')
 const RequestQueue = require('./index')
+const { KafkaJSCorrelationIdAlreadyExists } = require('../../errors')
 
 describe('Network > RequestQueue', () => {
   let requestQueue
@@ -187,6 +188,13 @@ describe('Network > RequestQueue', () => {
       const sentAt = await sendDone
       expect(sentAt).toBeGreaterThanOrEqual(before + clientSideThrottleTime)
     })
+
+    it('does not allow for a inflight correlation ids collision', async () => {
+      requestQueue.inflight.set(request.entry.correlationId, 'already existing inflight')
+      expect(() => {
+        requestQueue.push(request)
+      }).toThrowError(new KafkaJSCorrelationIdAlreadyExists(request.entry.correlationId))
+    })
   })
 
   describe('#fulfillRequest', () => {