Fix consumer not restarting on retriable connection errors

src/admin/__tests__/deleteTopicRecords.spec.js

@@ -339,9 +339,9 @@ describe('Admin > deleteTopicRecords', () => {
 
     expect(error).toBeDefined()
     expect(error.name).toBe('KafkaJSNonRetriableError')
-    expect(error.originalError.name).toBe('KafkaJSDeleteTopicRecordsError')
-    expect(error.originalError.retriable).toBe(false)
-    expect(error.originalError.partitions).toEqual([
+    expect(error.cause.name).toBe('KafkaJSDeleteTopicRecordsError')
+    expect(error.cause.retriable).toBe(false)
+    expect(error.cause.partitions).toEqual([
       {
         partition: 1,
         offset: '99',
@@ -403,9 +403,9 @@ describe('Admin > deleteTopicRecords', () => {
 
     expect(error).toBeDefined()
     expect(error.name).toBe('KafkaJSNonRetriableError')
-    expect(error.originalError.name).toBe('KafkaJSDeleteTopicRecordsError')
-    expect(error.originalError.retriable).toBe(false)
-    expect(error.originalError.partitions).toEqual(
+    expect(error.cause.name).toBe('KafkaJSDeleteTopicRecordsError')
+    expect(error.cause.retriable).toBe(false)
+    expect(error.cause.partitions).toEqual(
       expect.arrayContaining([
         {
           partition: 0,

src/consumer/__tests__/errorRecovery.spec.js

@@ -1,7 +1,11 @@
 const createProducer = require('../../producer')
 const createConsumer = require('../index')
 const { MemberMetadata, MemberAssignment } = require('../../consumer/assignerProtocol')
-const { KafkaJSError, KafkaJSNumberOfRetriesExceeded } = require('../../errors')
+const {
+  KafkaJSError,
+  KafkaJSNumberOfRetriesExceeded,
+  KafkaJSNonRetriableError,
+} = require('../../errors')
 
 const sleep = require('../../utils/sleep')
 const {
@@ -204,6 +208,58 @@ describe('Consumer', () => {
     })
   })
 
+  it('recovers from crashes due to retriable errors exhausting the number of retries', async () => {
+    const groupId = `consumer-group-id-${secureRandom()}`
+    consumer2 = createConsumer({
+      cluster,
+      groupId,
+      logger: newLogger(),
+      heartbeatInterval: 100,
+      maxWaitTimeInMs: 1,
+      maxBytesPerPartition: 180,
+      retry: {
+        retries: 0,
+      },
+    })
+    const crashListener = jest.fn()
+    consumer2.on(consumer.events.CRASH, crashListener)
+
+    const originalError = new KafkaJSError(new Error('💣'), { retriable: true })
+    const retryError = new KafkaJSNumberOfRetriesExceeded(originalError, {
+      retryCount: 5,
+      retryTime: 10000,
+      cause: originalError,
+    })
+    const error = new KafkaJSNonRetriableError(retryError, { cause: originalError })
+
+    await consumer2.connect()
+    await consumer2.subscribe({ topic: topicName, fromBeginning: true })
+
+    const coordinator = await cluster.findGroupCoordinator({ groupId })
+    const original = coordinator.joinGroup
+    coordinator.joinGroup = async () => {
+      coordinator.joinGroup = original
+      throw error
+    }
+
+    const eachMessage = jest.fn()
+    await consumer2.run({ eachMessage })
+
+    const key = secureRandom()
+    const message = { key: `key-${key}`, value: `value-${key}` }
+    await producer.send({ acks: 1, topic: topicName, messages: [message] })
+
+    await waitFor(() => crashListener.mock.calls.length > 0)
+    expect(crashListener).toHaveBeenCalledWith({
+      id: expect.any(Number),
+      timestamp: expect.any(Number),
+      type: 'consumer.crash',
+      payload: { error, groupId, restart: true },
+    })
+
+    await expect(waitFor(() => eachMessage.mock.calls.length)).resolves.toBe(1)
+  })
+
   it('recovers from retriable failures when "restartOnFailure" returns true', async () => {
     const errorMessage = '💣'
     let receivedErrorMessage

src/consumer/index.js

@@ -256,7 +256,16 @@ module.exports = ({
 
       await disconnect()
 
-      const isErrorRetriable = e.name === 'KafkaJSNumberOfRetriesExceeded' || e.retriable === true
+      const getOriginalCause = error => {
+        if (error.cause) {
+          return getOriginalCause(error.cause)
+        }
+
+        return error
+      }
+
+      const isErrorRetriable =
+        e.name === 'KafkaJSNumberOfRetriesExceeded' || getOriginalCause(e).retriable === true
       const shouldRestart =
         isErrorRetriable &&
         (!retry ||
@@ -266,7 +275,7 @@ module.exports = ({
               'Caught error when invoking user-provided "restartOnFailure" callback. Defaulting to restarting.',
               {
                 error: error.message || error,
-                originalError: e.message || e,
+                cause: e.message || e,
                 groupId,
               }
             )

src/errors.js

@@ -2,20 +2,22 @@ const pkgJson = require('../package.json')
 const { bugs } = pkgJson
 
 class KafkaJSError extends Error {
-  constructor(e, { retriable = true } = {}) {
-    super(e)
+  constructor(e, { retriable = true, cause } = {}) {
+    super(e, { cause })
     Error.captureStackTrace(this, this.constructor)
     this.message = e.message || e
     this.name = 'KafkaJSError'
     this.retriable = retriable
     this.helpUrl = e.helpUrl
+    this.cause = cause
   }
 }
 
 class KafkaJSNonRetriableError extends KafkaJSError {
-  constructor(e) {
-    super(e, { retriable: false })
+  constructor(e, { cause } = {}) {
+    super(e, { retriable: false, cause })
     this.name = 'KafkaJSNonRetriableError'
+    // Kept for backwards compatibility. Introduced in v1.16.0
     this.originalError = e
   }
 }
@@ -48,8 +50,9 @@ class KafkaJSMemberIdRequired extends KafkaJSProtocolError {
 
 class KafkaJSNumberOfRetriesExceeded extends KafkaJSNonRetriableError {
   constructor(e, { retryCount, retryTime }) {
-    super(e)
+    super(e, { cause: e })
     this.stack = `${this.name}\n  Caused by: ${e.stack}`
+    // Kept for backwards compatibility. Introduced in v1.16.0
     this.originalError = e
     this.retryCount = retryCount
     this.retryTime = retryTime

src/retry/index.js

@@ -47,10 +47,12 @@ const createRetriable = (configs, resolve, reject, fn) => {
           if (shouldRetry) {
             scheduleRetry()
           } else {
-            reject(new KafkaJSNumberOfRetriesExceeded(e, { retryCount, retryTime }))
+            reject(
+              new KafkaJSNumberOfRetriesExceeded(e, { retryCount, retryTime, cause: e.cause || e })
+            )
           }
         } else {
-          reject(new KafkaJSNonRetriableError(e))
+          reject(new KafkaJSNonRetriableError(e, { cause: e.cause || e }))
         }
       })
   }

types/index.d.ts

@@ -1039,6 +1039,7 @@ export class KafkaJSError extends Error {
   readonly name: string
   readonly retriable: boolean
   readonly helpUrl?: string
+  readonly cause?: Error
 
   constructor(e: Error | string, metadata?: KafkaJSErrorMetadata)
 }
@@ -1061,6 +1062,9 @@ export class KafkaJSOffsetOutOfRange extends KafkaJSProtocolError {
 
 export class KafkaJSNumberOfRetriesExceeded extends KafkaJSNonRetriableError {
   readonly stack: string
+  /**
+   * @deprecated Use `cause` instead
+   */
   readonly originalError: Error
   readonly retryCount: number
   readonly retryTime: number