Refresh metadata on any connection error

src/cluster/__tests__/brokerPool.spec.js

@@ -6,6 +6,7 @@ const {
   secureRandom,
 } = require('testHelpers')
 const { KafkaJSProtocolError, KafkaJSConnectionError } = require('../../errors')
+const { createErrorFromCode, errorCodes } = require('../../protocol/error')
 const BrokerPool = require('../brokerPool')
 const Broker = require('../../broker')
 
@@ -393,14 +394,14 @@ describe('Cluster > BrokerPool', () => {
       expect(broker.isConnected()).toEqual(true)
     })
 
-    it('recreates the connection on connection errors', async () => {
+    it('recreates the connection on ILLEGAL_SASL_STATE error', async () => {
       const nodeId = 'fakebroker'
       const mockBroker = new Broker({
         connection: createConnection(),
         logger: newLogger(),
       })
       jest.spyOn(mockBroker, 'connect').mockImplementationOnce(() => {
-        throw new KafkaJSConnectionError('Connection lost')
+        throw createErrorFromCode(errorCodes.find(({ type }) => type === 'ILLEGAL_SASL_STATE').code)
       })
       brokerPool.brokers[nodeId] = mockBroker
 

src/cluster/__tests__/findBroker.spec.js

@@ -97,4 +97,27 @@ describe('Cluster > findBroker', () => {
 
     expect(cluster.refreshMetadata).toHaveBeenCalled()
   })
+
+  test('refresh metadata on KafkaJSConnectionError Connection Timeout', async () => {
+    const nodeId = 0
+    const mockBroker = new Broker({
+      connection: createConnection(),
+      logger: newLogger(),
+    })
+
+    jest.spyOn(mockBroker, 'connect').mockImplementationOnce(() => {
+      throw new KafkaJSConnectionError('Connection timeout', { broker: mockBroker })
+    })
+
+    jest.spyOn(cluster, 'refreshMetadata')
+    cluster.brokerPool.brokers[nodeId] = mockBroker
+
+    await expect(cluster.findBroker({ nodeId })).rejects.toHaveProperty(
+      'name',
+      'KafkaJSConnectionError'
+    )
+
+    await expect(cluster.findBroker({ nodeId })).resolves.toBeInstanceOf(Broker)
+    expect(cluster.refreshMetadata).toHaveBeenCalled()
+  })
 })

src/cluster/brokerPool.js

@@ -2,7 +2,7 @@ const Broker = require('../broker')
 const createRetry = require('../retry')
 const shuffle = require('../utils/shuffle')
 const arrayDiff = require('../utils/arrayDiff')
-const { KafkaJSBrokerNotFound } = require('../errors')
+const { KafkaJSBrokerNotFound, KafkaJSProtocolError } = require('../errors')
 
 const { keys, assign, values } = Object
 const hasBrokerBeenReplaced = (broker, { host, port, rack }) =>
@@ -311,13 +311,15 @@ module.exports = class BrokerPool {
       } catch (e) {
         if (e.name === 'KafkaJSConnectionError' || e.type === 'ILLEGAL_SASL_STATE') {
           await broker.disconnect()
+        }
 
-          // Connection refused means this node is down, or the cluster is restarting,
-          // which requires metadata refresh to discover the new nodes
-          if (e.code === 'ECONNREFUSED') {
-            return bail(e)
-          }
+        // To avoid reconnecting to an unavailable host, we bail on connection errors
+        // and refresh metadata on a higher level before reconnecting
+        if (e.name === 'KafkaJSConnectionError') {
+          return bail(e)
+        }
 
+        if (e.type === 'ILLEGAL_SASL_STATE') {
           // Rebuild the connection since it can't recover from illegal SASL state
           broker.connection = await this.connectionBuilder.build({
             host: broker.connection.host,
@@ -326,6 +328,7 @@ module.exports = class BrokerPool {
           })
 
           this.logger.error(`Failed to connect to broker, reconnecting`, { retryCount, retryTime })
+          throw new KafkaJSProtocolError(e, { retriable: true })
         }
 
         if (e.retriable) throw e

src/cluster/index.js

@@ -215,7 +215,7 @@ module.exports = class Cluster {
       if (
         e.name === 'KafkaJSBrokerNotFound' ||
         e.name === 'KafkaJSLockTimeout' ||
-        e.code === 'ECONNREFUSED'
+        e.name === 'KafkaJSConnectionError'
       ) {
         await this.refreshMetadata()
       }

src/errors.js

@@ -19,8 +19,8 @@ class KafkaJSNonRetriableError extends KafkaJSError {
 }
 
 class KafkaJSProtocolError extends KafkaJSError {
-  constructor(e) {
-    super(e, { retriable: e.retriable })
+  constructor(e, { retriable = e.retriable } = {}) {
+    super(e, { retriable })
     this.type = e.type
     this.code = e.code
     this.name = 'KafkaJSProtocolError'