Add single requestTimeout runner instead of setTimeout per request

src/network/connection.js

@@ -75,6 +75,7 @@ module.exports = class Connection {
       clientId,
       broker: this.broker,
       logger: logger.namespace('RequestQueue'),
+      isConnected: () => this.connected,
     })
 
     this.authHandlers = null
@@ -109,6 +110,7 @@ module.exports = class Connection {
       const onConnect = () => {
         clearTimeout(timeoutId)
         this.connected = true
+        this.requestQueue.scheduleRequestTimeoutCheck()
         resolve(true)
       }
 
@@ -199,6 +201,7 @@ module.exports = class Connection {
     }
 
     this.logDebug('disconnecting...')
+    this.requestQueue.destroy()
     this.connected = false
     this.socket.end()
     this.socket.unref()

src/network/connection.spec.js

@@ -141,7 +141,10 @@ describe('Network > Connection', () => {
     test('respect the requestTimeout', async () => {
       const protocol = apiVersions()
       connection = new Connection(
-        connectionOpts({ requestTimeout: 50, enforceRequestTimeout: true })
+        connectionOpts({
+          requestTimeout: 50,
+          enforceRequestTimeout: true,
+        })
       )
       const originalProcessData = connection.processData
 

src/network/requestQueue/index.js

@@ -22,6 +22,7 @@ module.exports = class RequestQueue {
     clientId,
     broker,
     logger,
+    isConnected = () => true,
   }) {
     this.instrumentationEmitter = instrumentationEmitter
     this.maxInFlightRequests = maxInFlightRequests
@@ -30,6 +31,7 @@ module.exports = class RequestQueue {
     this.clientId = clientId
     this.broker = broker
     this.logger = logger
+    this.isConnected = isConnected
 
     this.inflight = new Map()
     this.pending = []
@@ -44,6 +46,27 @@ module.exports = class RequestQueue {
     }
   }
 
+  /**
+   * @public
+   */
+  scheduleRequestTimeoutCheck() {
+    if (this.enforceRequestTimeout) {
+      this.destroy()
+
+      this.requestTimeoutIntervalId = setInterval(() => {
+        this.inflight.forEach(request => {
+          if (Date.now() - request.sentAt > request.requestTimeout) {
+            request.timeoutRequest()
+          }
+
+          if (!this.isConnected()) {
+            this.destroy()
+          }
+        })
+      }, Math.min(this.requestTimeout, 100))
+    }
+  }
+
   /**
    * @typedef {Object} PushedRequest
    * @property {RequestEntry} entry
@@ -69,7 +92,6 @@ module.exports = class RequestQueue {
       broker: this.broker,
       clientId: this.clientId,
       instrumentationEmitter: this.instrumentationEmitter,
-      enforceRequestTimeout: this.enforceRequestTimeout,
       requestTimeout,
       send: () => {
         this.inflight.set(correlationId, socketRequest)
@@ -169,4 +191,11 @@ module.exports = class RequestQueue {
     this.inflight.clear()
     this[PRIVATE.EMIT_QUEUE_SIZE_EVENT]()
   }
+
+  /**
+   * @public
+   */
+  destroy() {
+    clearInterval(this.requestTimeoutIntervalId)
+  }
 }

src/network/requestQueue/index.spec.js

@@ -1,3 +1,4 @@
+const sleep = require('../../utils/sleep')
 const { newLogger } = require('testHelpers')
 const InstrumentationEventEmitter = require('../../instrumentation/emitter')
 const events = require('../instrumentationEvents')
@@ -196,12 +197,17 @@ describe('Network > RequestQueue', () => {
   })
 
   describe('instrumentation events', () => {
-    let emitter, removeListener, eventCalled, request, payload, size
+    let emitter, removeListener, eventCalled, request, payload, size, requestTimeout
 
     beforeEach(() => {
+      requestTimeout = 1
       eventCalled = jest.fn()
       emitter = new InstrumentationEventEmitter()
-      requestQueue = createRequestQueue({ instrumentationEmitter: emitter })
+      requestQueue = createRequestQueue({
+        instrumentationEmitter: emitter,
+        enforceRequestTimeout: true,
+        requestTimeout,
+      })
       request = {
         sendRequest: jest.fn(),
         entry: createEntry(),
@@ -298,5 +304,30 @@ describe('Network > RequestQueue', () => {
         },
       })
     })
+
+    it('emits NETWORK_REQUEST_TIMEOUT', async () => {
+      emitter.addListener(events.NETWORK_REQUEST_TIMEOUT, eventCalled)
+      requestQueue.scheduleRequestTimeoutCheck()
+      requestQueue.push(request)
+
+      await sleep(requestTimeout + 1)
+
+      expect(eventCalled).toHaveBeenCalledWith({
+        id: expect.any(Number),
+        type: 'network.request_timeout',
+        timestamp: expect.any(Number),
+        payload: {
+          apiKey: request.entry.apiKey,
+          apiName: request.entry.apiName,
+          apiVersion: request.entry.apiVersion,
+          broker: 'localhost:9092',
+          clientId: 'KafkaJS',
+          correlationId: expect.any(Number),
+          createdAt: expect.any(Number),
+          pendingDuration: expect.any(Number),
+          sentAt: expect.any(Number),
+        },
+      })
+    })
   })
 })

src/network/requestQueue/socketRequest.js

@@ -49,7 +49,6 @@ module.exports = class SocketRequest {
    */
   constructor({
     requestTimeout,
-    enforceRequestTimeout,
     broker,
     clientId,
     entry,
@@ -60,7 +59,6 @@ module.exports = class SocketRequest {
   }) {
     this.createdAt = Date.now()
     this.requestTimeout = requestTimeout
-    this.enforceRequestTimeout = enforceRequestTimeout
     this.broker = broker
     this.clientId = clientId
     this.entry = entry
@@ -72,7 +70,6 @@ module.exports = class SocketRequest {
     this.sentAt = null
     this.duration = null
     this.pendingDuration = null
-    this.timeoutId = null
 
     this[PRIVATE.STATE] = REQUEST_STATE.PENDING
     this[PRIVATE.EMIT_EVENT] = (eventName, payload) =>
@@ -89,32 +86,28 @@ module.exports = class SocketRequest {
     this.sentAt = Date.now()
     this.pendingDuration = this.sentAt - this.createdAt
     this[PRIVATE.STATE] = REQUEST_STATE.SENT
+  }
 
-    const timeoutCallback = () => {
-      const { apiName, apiKey, apiVersion } = this.entry
-      const requestInfo = `${apiName}(key: ${apiKey}, version: ${apiVersion})`
-      const eventData = {
-        broker: this.broker,
-        clientId: this.clientId,
-        correlationId: this.correlationId,
-        createdAt: this.createdAt,
-        sentAt: this.sentAt,
-        pendingDuration: this.pendingDuration,
-      }
-
-      this.timeoutHandler()
-      this.rejected(new KafkaJSRequestTimeoutError(`Request ${requestInfo} timed out`, eventData))
-      this[PRIVATE.EMIT_EVENT](events.NETWORK_REQUEST_TIMEOUT, {
-        ...eventData,
-        apiName,
-        apiKey,
-        apiVersion,
-      })
+  timeoutRequest() {
+    const { apiName, apiKey, apiVersion } = this.entry
+    const requestInfo = `${apiName}(key: ${apiKey}, version: ${apiVersion})`
+    const eventData = {
+      broker: this.broker,
+      clientId: this.clientId,
+      correlationId: this.correlationId,
+      createdAt: this.createdAt,
+      sentAt: this.sentAt,
+      pendingDuration: this.pendingDuration,
     }
 
-    if (this.enforceRequestTimeout) {
-      this.timeoutId = setTimeout(timeoutCallback, this.requestTimeout)
-    }
+    this.timeoutHandler()
+    this.rejected(new KafkaJSRequestTimeoutError(`Request ${requestInfo} timed out`, eventData))
+    this[PRIVATE.EMIT_EVENT](events.NETWORK_REQUEST_TIMEOUT, {
+      ...eventData,
+      apiName,
+      apiKey,
+      apiVersion,
+    })
   }
 
   completed({ size, payload }) {
@@ -123,7 +116,6 @@ module.exports = class SocketRequest {
       next: REQUEST_STATE.COMPLETED,
     })
 
-    clearTimeout(this.timeoutId)
     const { entry, correlationId, broker, clientId, createdAt, sentAt, pendingDuration } = this
 
     this[PRIVATE.STATE] = REQUEST_STATE.COMPLETED
@@ -151,7 +143,6 @@ module.exports = class SocketRequest {
       next: REQUEST_STATE.REJECTED,
     })
 
-    clearTimeout(this.timeoutId)
     this[PRIVATE.STATE] = REQUEST_STATE.REJECTED
     this.duration = Date.now() - this.sentAt
     this.entry.reject(error)

src/network/requestQueue/socketRequest.spec.js

@@ -1,4 +1,3 @@
-const sleep = require('../../utils/sleep')
 const SocketRequest = require('./socketRequest')
 const { KafkaJSRequestTimeoutError, KafkaJSNonRetriableError } = require('../../errors')
 const InstrumentationEventEmitter = require('../../instrumentation/emitter')
@@ -41,33 +40,28 @@ describe('Network > SocketRequest', () => {
     it('sends the request using the provided function', () => {
       expect(request.sentAt).toEqual(null)
       expect(request.pendingDuration).toEqual(null)
-      expect(request.timeoutId).toEqual(null)
 
       request.enforceRequestTimeout = true
       request.send()
 
       expect(sendRequest).toHaveBeenCalled()
       expect(request.sentAt).toEqual(expect.any(Number))
       expect(request.pendingDuration).toEqual(expect.any(Number))
-      expect(request.timeoutId).not.toEqual(null)
-
-      clearTimeout(request.timeoutId)
     })
 
     it('does not call sendRequest more than once', () => {
       request.send()
       expect(() => request.send()).toThrow(KafkaJSNonRetriableError)
 
       expect(sendRequest).toHaveBeenCalledTimes(1)
-      clearTimeout(request.timeoutId)
     })
 
     it('executes the timeoutHandler when it times out', async () => {
       jest.spyOn(request, 'rejected')
       request.enforceRequestTimeout = true
       request.send()
+      request.timeoutRequest()
 
-      await sleep(requestTimeout + 1)
       expect(request.rejected).toHaveBeenCalled()
       expect(request.entry.reject).toHaveBeenCalledWith(expect.any(KafkaJSRequestTimeoutError))
       expect(timeoutHandler).toHaveBeenCalled()
@@ -162,32 +156,5 @@ describe('Network > SocketRequest', () => {
         },
       })
     })
-
-    it('emits NETWORK_REQUEST_TIMEOUT', async () => {
-      jest.spyOn(request, 'rejected')
-      emitter.addListener(events.NETWORK_REQUEST_TIMEOUT, eventCalled)
-      request.enforceRequestTimeout = true
-      request.send()
-
-      await sleep(requestTimeout + 1)
-
-      expect(timeoutHandler).toHaveBeenCalled()
-      expect(eventCalled).toHaveBeenCalledWith({
-        id: expect.any(Number),
-        type: 'network.request_timeout',
-        timestamp: expect.any(Number),
-        payload: {
-          apiKey: 0,
-          apiName: 'Produce',
-          apiVersion: 4,
-          broker: 'localhost:9092',
-          clientId: 'KafkaJS',
-          correlationId: expect.any(Number),
-          createdAt: expect.any(Number),
-          pendingDuration: expect.any(Number),
-          sentAt: expect.any(Number),
-        },
-      })
-    })
   })
 })