#683 Improve concurrency

index.js

@@ -10,7 +10,7 @@ const AclResourceTypes = require('./src/protocol/aclResourceTypes')
 const AclOperationTypes = require('./src/protocol/aclOperationTypes')
 const AclPermissionTypes = require('./src/protocol/aclPermissionTypes')
 const ResourcePatternTypes = require('./src/protocol/resourcePatternTypes')
-const Errors = require('./src/errors')
+const { isRebalancing, isKafkaJSError, ...errors } = require('./src/errors')
 const { LEVELS } = require('./src/loggers')
 
 module.exports = {
@@ -34,5 +34,5 @@ module.exports = {
   AclPermissionTypes,
   ResourcePatternTypes,
   ConfigSource,
-  ...Errors,
+  ...errors,
 }

package.json

@@ -46,6 +46,7 @@
     "test:types": "tsc -p types/"
   },
   "devDependencies": {
+    "@types/jest": "^27.4.0",
     "@types/node": "^12.0.8",
     "@typescript-eslint/typescript-estree": "^1.10.2",
     "eslint": "^6.8.0",

src/admin/__tests__/fetchTopicOffsetsByTimestamp.spec.js

@@ -13,7 +13,7 @@ const {
 } = require('testHelpers')
 
 describe('Admin', () => {
-  let topicName, admin, producer, cluster
+  let topicName, admin, producer, cluster, consumer
 
   beforeEach(async () => {
     topicName = `test-topic-${secureRandom()}`
@@ -30,6 +30,7 @@ describe('Admin', () => {
   })
   afterEach(async () => {
     admin && (await admin.disconnect())
+    consumer && (await consumer.disconnect())
     producer && (await producer.disconnect())
   })
 
@@ -72,7 +73,7 @@ describe('Admin', () => {
       )
       expect(offsetsFutureTimestamp).toEqual([{ partition: 0, offset: '20' }])
       const groupId = `consumer-group-id-${secureRandom()}`
-      const consumer = createConsumer({
+      consumer = createConsumer({
         cluster,
         groupId,
         maxWaitTimeInMs: 1,

src/admin/index.spec.js

@@ -4,6 +4,12 @@ const { createCluster, newLogger, secureRandom } = require('testHelpers')
 const createRetry = require('../retry')
 
 describe('Admin', () => {
+  let admin
+
+  afterEach(async () => {
+    admin && (await admin.disconnect())
+  })
+
   it('gives access to its logger', () => {
     expect(
       createAdmin({
@@ -14,7 +20,7 @@ describe('Admin', () => {
   })
 
   it('emits connection events', async () => {
-    const admin = createAdmin({
+    admin = createAdmin({
       cluster: createCluster(),
       logger: newLogger(),
     })
@@ -33,7 +39,7 @@ describe('Admin', () => {
 
   test('emits the request event', async () => {
     const emitter = new InstrumentationEventEmitter()
-    const admin = createAdmin({
+    admin = createAdmin({
       cluster: createCluster({ instrumentationEmitter: emitter }),
       logger: newLogger(),
       instrumentationEmitter: emitter,
@@ -72,7 +78,7 @@ describe('Admin', () => {
       instrumentationEmitter: emitter,
     })
 
-    const admin = createAdmin({
+    admin = createAdmin({
       cluster,
       logger: newLogger(),
       instrumentationEmitter: emitter,
@@ -121,7 +127,7 @@ describe('Admin', () => {
       maxInFlightRequests: 1,
     })
 
-    const admin = createAdmin({
+    admin = createAdmin({
       cluster,
       logger: newLogger(),
       instrumentationEmitter: emitter,
@@ -155,7 +161,7 @@ describe('Admin', () => {
   })
 
   test('on throws an error when provided with an invalid event name', () => {
-    const admin = createAdmin({
+    admin = createAdmin({
       cluster: createCluster(),
       logger: newLogger(),
     })

src/broker/__tests__/addOffsetsToTxn.spec.js

@@ -1,6 +1,6 @@
 const Broker = require('../index')
 const COORDINATOR_TYPES = require('../../protocol/coordinatorTypes')
-const { secureRandom, createConnection, newLogger, retryProtocol } = require('testHelpers')
+const { secureRandom, createConnectionPool, newLogger, retryProtocol } = require('testHelpers')
 const { KafkaJSProtocolError } = require('../../errors')
 
 describe('Broker > AddOffsetsToTxn', () => {
@@ -11,7 +11,7 @@ describe('Broker > AddOffsetsToTxn', () => {
     consumerGroupId = `group-id-${secureRandom()}`
 
     seedBroker = new Broker({
-      connection: createConnection(),
+      connectionPool: createConnectionPool(),
       logger: newLogger(),
     })
 
@@ -29,7 +29,7 @@ describe('Broker > AddOffsetsToTxn', () => {
     )
 
     broker = new Broker({
-      connection: createConnection({ host, port }),
+      connectionPool: createConnectionPool({ host, port }),
       logger: newLogger(),
     })
 

src/broker/__tests__/addPartitionsToTxn.spec.js

@@ -4,7 +4,7 @@ const { KafkaJSProtocolError } = require('../../errors')
 const {
   secureRandom,
   createTopic,
-  createConnection,
+  createConnectionPool,
   newLogger,
   retryProtocol,
 } = require('testHelpers')
@@ -17,7 +17,7 @@ describe('Broker > AddPartitionsToTxn', () => {
     topicName = `test-topic-${secureRandom()}`
 
     seedBroker = new Broker({
-      connection: createConnection(),
+      connectionPool: createConnectionPool(),
       logger: newLogger(),
     })
 
@@ -36,7 +36,7 @@ describe('Broker > AddPartitionsToTxn', () => {
     )
 
     broker = new Broker({
-      connection: createConnection({ host, port }),
+      connectionPool: createConnectionPool({ host, port }),
       logger: newLogger(),
     })
 

src/broker/__tests__/alterConfigs.spec.js

@@ -1,4 +1,4 @@
-const { createConnection, connectionOpts, secureRandom, newLogger } = require('testHelpers')
+const { createConnectionPool, connectionOpts, secureRandom, newLogger } = require('testHelpers')
 const RESOURCE_TYPES = require('../../protocol/resourceTypes')
 const Broker = require('../index')
 
@@ -13,7 +13,7 @@ describe('Broker > alterConfigs', () => {
 
   beforeEach(async () => {
     seedBroker = new Broker({
-      connection: createConnection(connectionOpts()),
+      connectionPool: createConnectionPool(connectionOpts()),
       logger: newLogger(),
     })
     await seedBroker.connect()
@@ -22,7 +22,7 @@ describe('Broker > alterConfigs', () => {
     const newBrokerData = metadata.brokers.find(b => b.nodeId === metadata.controllerId)
 
     broker = new Broker({
-      connection: createConnection(newBrokerData),
+      connectionPool: createConnectionPool(newBrokerData),
       logger: newLogger(),
     })
   })

src/broker/__tests__/apiVersions.spec.js

@@ -1,4 +1,4 @@
-const { createConnection, connectionOpts, newLogger } = require('testHelpers')
+const { createConnectionPool, connectionOpts, newLogger } = require('testHelpers')
 const { KafkaJSProtocolError, KafkaJSNonRetriableError } = require('../../errors')
 const { createErrorFromCode } = require('../../protocol/error')
 const { requests } = require('../../protocol/requests')
@@ -11,7 +11,7 @@ describe('Broker > ApiVersions', () => {
 
   beforeEach(async () => {
     broker = new Broker({
-      connection: createConnection(connectionOpts()),
+      connectionPool: createConnectionPool(connectionOpts()),
       logger: newLogger(),
     })
     await broker.connect()

src/broker/__tests__/connect.spec.js

@@ -1,5 +1,5 @@
 const {
-  createConnection,
+  createConnectionPool,
   connectionOpts,
   saslSCRAM256ConnectionOpts,
   newLogger,
@@ -12,13 +12,11 @@ const Long = require('../../utils/long')
 const Broker = require('../index')
 
 describe('Broker > connect', () => {
-  let broker
+  let broker, connectionPool
 
   beforeEach(() => {
-    broker = new Broker({
-      connection: createConnection(connectionOpts()),
-      logger: newLogger(),
-    })
+    connectionPool = createConnectionPool(connectionOpts())
+    broker = new Broker({ connectionPool, logger: newLogger() })
   })
 
   afterEach(async () => {
@@ -27,7 +25,7 @@ describe('Broker > connect', () => {
 
   test('establish the connection', async () => {
     await broker.connect()
-    expect(broker.connection.connected).toEqual(true)
+    expect(broker.connectionPool.isConnected()).toEqual(true)
   })
 
   test('load api versions if not provided', async () => {
@@ -39,7 +37,7 @@ describe('Broker > connect', () => {
   for (const e of saslEntries) {
     test(`authenticate with SASL ${e.name} if configured`, async () => {
       broker = new Broker({
-        connection: createConnection(e.opts()),
+        connectionPool: createConnectionPool(e.opts()),
         logger: newLogger(),
       })
       expect(broker.isConnected()).toEqual(false)
@@ -51,7 +49,7 @@ describe('Broker > connect', () => {
   describeIfOauthbearerDisabled('when SASL SCRAM is configured', () => {
     test('parallel calls to connect using SCRAM', async () => {
       broker = new Broker({
-        connection: createConnection(saslSCRAM256ConnectionOpts()),
+        connectionPool: createConnectionPool(saslSCRAM256ConnectionOpts()),
         logger: newLogger(),
       })
 
@@ -69,32 +67,17 @@ describe('Broker > connect', () => {
     })
   })
 
-  test('sets the authenticatedAt timer', async () => {
-    const error = new Error('not connected')
-    const timer = process.hrtime()
-    broker.authenticatedAt = timer
-    broker.connection.connect = jest.fn(() => {
-      throw error
-    })
-
-    expect(broker.authenticatedAt).toEqual(timer)
-    await expect(broker.connect()).rejects.toEqual(error)
-    expect(broker.authenticatedAt).toBe(null)
-  })
-
   describe('#isConnected', () => {
     test('returns false when not connected', () => {
       expect(broker.isConnected()).toEqual(false)
     })
 
     for (const e of saslEntries) {
       test(`returns false when connected but not authenticated on connections with SASL ${e.name}`, async () => {
-        broker = new Broker({
-          connection: createConnection(e.opts()),
-          logger: newLogger(),
-        })
+        const connectionPool = createConnectionPool(e.opts())
+        broker = new Broker({ connectionPool, logger: newLogger() })
         expect(broker.isConnected()).toEqual(false)
-        await broker.connection.connect()
+        await connectionPool.getConnection()
         expect(broker.isConnected()).toEqual(false)
       })
     }
@@ -107,65 +90,61 @@ describe('Broker > connect', () => {
     describe('when SaslAuthenticate protocol is available', () => {
       for (const e of saslEntries) {
         test(`returns true when connected and authenticated on connections with SASL ${e.name}`, async () => {
-          broker = new Broker({
-            connection: createConnection(e.opts()),
-            logger: newLogger(),
-          })
+          const connectionPool = createConnectionPool(e.opts())
+          broker = new Broker({ connectionPool, logger: newLogger() })
           await broker.connect()
           expect(broker.isConnected()).toEqual(true)
         })
 
         test('returns false when the session lifetime has expired', async () => {
           const sessionLifetime = 15000
           const reauthenticationThreshold = 10000
-          broker = new Broker({
-            connection: createConnection(e.opts()),
-            logger: newLogger(),
-            reauthenticationThreshold,
-          })
+          const connectionPool = createConnectionPool({ ...e.opts(), reauthenticationThreshold })
+          broker = new Broker({ connectionPool, logger: newLogger() })
 
           await broker.connect()
           expect(broker.isConnected()).toEqual(true)
 
-          broker.sessionLifetime = Long.fromValue(sessionLifetime)
-          const [seconds] = broker.authenticatedAt
-          broker.authenticatedAt = [seconds - sessionLifetime / 1000, 0]
+          const connection = await connectionPool.getConnection()
+
+          connection.sessionLifetime = Long.fromValue(sessionLifetime)
+          const [seconds] = connection.authenticatedAt
+          connection.authenticatedAt = [seconds - sessionLifetime / 1000, 0]
 
           expect(broker.isConnected()).toEqual(false)
         })
 
         test('returns true when the session lifetime is 0', async () => {
-          broker = new Broker({
-            connection: createConnection(e.opts()),
-            logger: newLogger(),
-          })
+          const connectionPool = createConnectionPool(e.opts())
+          broker = new Broker({ connectionPool, logger: newLogger() })
 
           await broker.connect()
           expect(broker.isConnected()).toEqual(true)
 
-          broker.sessionLifetime = Long.ZERO
-          broker.authenticatedAt = [0, 0]
+          const connection = await connectionPool.getConnection()
+          connection.sessionLifetime = Long.ZERO
+          connection.authenticatedAt = [0, 0]
 
           expect(broker.isConnected()).toEqual(true)
         })
 
         testIfKafkaAtLeast_1_1_0(`authenticate with SASL ${e.name} if configured`, async () => {
-          broker = new Broker({
-            connection: createConnection(e.opts()),
-            logger: newLogger(),
-          })
+          const connectionPool = createConnectionPool(e.opts())
+          broker = new Broker({ connectionPool, logger: newLogger() })
           expect(broker.isConnected()).toEqual(false)
           await broker.connect()
           expect(broker.isConnected()).toEqual(true)
-          expect(broker.supportAuthenticationProtocol).toEqual(true)
+
+          const connection = await connectionPool.getConnection()
+          expect(connection.getSupportAuthenticationProtocol()).toEqual(true)
         })
       }
     })
 
     describeIfOauthbearerDisabled('when SASL SCRAM is configured', () => {
       testIfKafkaAtLeast_1_1_0('parallel calls to connect using SCRAM', async () => {
         broker = new Broker({
-          connection: createConnection(saslSCRAM256ConnectionOpts()),
+          connectionPool: createConnectionPool(saslSCRAM256ConnectionOpts()),
           logger: newLogger(),
         })
 

src/broker/__tests__/createAcls.spec.js

@@ -1,4 +1,4 @@
-const { createConnection, connectionOpts, secureRandom, newLogger } = require('testHelpers')
+const { createConnectionPool, connectionOpts, secureRandom, newLogger } = require('testHelpers')
 
 const Broker = require('../index')
 const ACL_RESOURCE_TYPES = require('../../protocol/aclResourceTypes')
@@ -11,7 +11,7 @@ describe('Broker > createAcls', () => {
 
   beforeEach(async () => {
     seedBroker = new Broker({
-      connection: createConnection(connectionOpts()),
+      connectionPool: createConnectionPool(connectionOpts()),
       logger: newLogger(),
     })
     await seedBroker.connect()
@@ -20,7 +20,7 @@ describe('Broker > createAcls', () => {
     const newBrokerData = metadata.brokers.find(b => b.nodeId === metadata.controllerId)
 
     broker = new Broker({
-      connection: createConnection(newBrokerData),
+      connectionPool: createConnectionPool(newBrokerData),
       logger: newLogger(),
     })
   })