Merge pull request #313 from tulios/make-request-timeout-optional

Make the requestTimeout optional

src/admin/index.spec.js

@@ -66,6 +66,7 @@ describe('Admin', () => {
     const emitter = new InstrumentationEventEmitter()
     const cluster = createCluster({
       requestTimeout: 1,
+      enforceRequestTimeout: true,
       instrumentationEmitter: emitter,
     })
 

src/cluster/connectionBuilder.js

@@ -6,6 +6,7 @@ module.exports = ({
   sasl,
   clientId,
   requestTimeout,
+  enforceRequestTimeout,
   connectionTimeout,
   maxInFlightRequests,
   retry,
@@ -33,6 +34,7 @@ module.exports = ({
         clientId,
         connectionTimeout,
         requestTimeout,
+        enforceRequestTimeout,
         maxInFlightRequests,
         instrumentationEmitter,
         retry,

src/cluster/index.js

@@ -46,6 +46,7 @@ module.exports = class Cluster {
     connectionTimeout,
     authenticationTimeout,
     requestTimeout,
+    enforceRequestTimeout,
     metadataMaxAge,
     retry,
     allowExperimentalV011,
@@ -67,6 +68,7 @@ module.exports = class Cluster {
       clientId,
       connectionTimeout,
       requestTimeout,
+      enforceRequestTimeout,
       maxInFlightRequests,
       retry,
     })

src/consumer/__tests__/instrumentationEvents.spec.js

@@ -315,7 +315,11 @@ describe('Consumer > Instrumentation Events', () => {
   })
 
   it('emits request timeout events', async () => {
-    cluster = createCluster({ instrumentationEmitter: emitter, requestTimeout: 1 })
+    cluster = createCluster({
+      instrumentationEmitter: emitter,
+      requestTimeout: 1,
+      enforceRequestTimeout: true,
+    })
     consumer = createConsumer({
       cluster,
       groupId,

src/index.js

@@ -27,6 +27,7 @@ module.exports = class Client {
     connectionTimeout,
     authenticationTimeout,
     requestTimeout,
+    enforceRequestTimeout = false,
     retry,
     logLevel = INFO,
     logCreator = LoggerConsole,
@@ -52,6 +53,7 @@ module.exports = class Client {
         connectionTimeout,
         authenticationTimeout,
         requestTimeout,
+        enforceRequestTimeout,
         metadataMaxAge,
         instrumentationEmitter,
         allowAutoTopicCreation,

src/network/connection.js

@@ -42,6 +42,7 @@ module.exports = class Connection {
     clientId = 'kafkajs',
     connectionTimeout = 1000,
     requestTimeout = 30000,
+    enforceRequestTimeout = false,
     maxInFlightRequests = null,
     instrumentationEmitter = null,
     retry = {},
@@ -68,6 +69,7 @@ module.exports = class Connection {
       instrumentationEmitter,
       maxInFlightRequests,
       requestTimeout,
+      enforceRequestTimeout,
       clientId,
       broker: this.broker,
       logger: logger.namespace('RequestQueue'),

src/network/connection.spec.js

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

src/network/requestQueue/index.js

@@ -18,13 +18,15 @@ module.exports = class RequestQueue {
     instrumentationEmitter = null,
     maxInFlightRequests,
     requestTimeout,
+    enforceRequestTimeout,
     clientId,
     broker,
     logger,
   }) {
     this.instrumentationEmitter = instrumentationEmitter
     this.maxInFlightRequests = maxInFlightRequests
     this.requestTimeout = requestTimeout
+    this.enforceRequestTimeout = enforceRequestTimeout
     this.clientId = clientId
     this.broker = broker
     this.logger = logger
@@ -65,6 +67,7 @@ module.exports = class RequestQueue {
       broker: this.broker,
       clientId: this.clientId,
       instrumentationEmitter: this.instrumentationEmitter,
+      enforceRequestTimeout: this.enforceRequestTimeout,
       requestTimeout,
       send: () => {
         this.inflight.set(correlationId, socketRequest)

src/network/requestQueue/socketRequest.js

@@ -49,6 +49,7 @@ module.exports = class SocketRequest {
    */
   constructor({
     requestTimeout,
+    enforceRequestTimeout,
     broker,
     clientId,
     entry,
@@ -59,6 +60,7 @@ module.exports = class SocketRequest {
   }) {
     this.createdAt = Date.now()
     this.requestTimeout = requestTimeout
+    this.enforceRequestTimeout = enforceRequestTimeout
     this.broker = broker
     this.clientId = clientId
     this.entry = entry
@@ -110,7 +112,9 @@ module.exports = class SocketRequest {
       })
     }
 
-    this.timeoutId = setTimeout(timeoutCallback, this.requestTimeout)
+    if (this.enforceRequestTimeout) {
+      this.timeoutId = setTimeout(timeoutCallback, this.requestTimeout)
+    }
   }
 
   completed({ size, payload }) {

src/network/requestQueue/socketRequest.spec.js

@@ -43,6 +43,7 @@ describe('Network > SocketRequest', () => {
       expect(request.pendingDuration).toEqual(null)
       expect(request.timeoutId).toEqual(null)
 
+      request.enforceRequestTimeout = true
       request.send()
 
       expect(sendRequest).toHaveBeenCalled()
@@ -63,6 +64,7 @@ describe('Network > SocketRequest', () => {
 
     it('executes the timeoutHandler when it times out', async () => {
       jest.spyOn(request, 'rejected')
+      request.enforceRequestTimeout = true
       request.send()
 
       await sleep(requestTimeout + 1)
@@ -164,6 +166,7 @@ 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)

src/producer/index.spec.js

@@ -255,6 +255,7 @@ describe('Producer', () => {
     const emitter = new InstrumentationEventEmitter()
     const cluster = createCluster({
       requestTimeout: 1,
+      enforceRequestTimeout: true,
       instrumentationEmitter: emitter,
     })