Create delete topic error structure

src/admin/__tests__/deleteTopics.spec.js

@@ -48,6 +48,16 @@ describe('Admin', () => {
       await expect(admin.deleteTopics({ topics: [topicName] })).resolves.toBe()
     })
 
+    test('failt to delete not existing topic', async () => {
+      admin = createAdmin({ cluster: createCluster(), logger: newLogger() })
+
+      await admin.connect()
+      await expect(admin.deleteTopics({ topics: [topicName] })).rejects.toHaveProperty(
+        'message',
+        'Delete topics error'
+      )
+    })
+
     test('remove deleted topics from the cluster target group', async () => {
       const cluster = createCluster()
       admin = createAdmin({ cluster, logger: newLogger() })

src/admin/index.js

@@ -270,8 +270,25 @@ module.exports = ({
 
         await cluster.refreshMetadata()
       } catch (e) {
-        if (['NOT_CONTROLLER', 'UNKNOWN_TOPIC_OR_PARTITION'].includes(e.type)) {
-          logger.warn('Could not delete topics', { error: e.message, retryCount, retryTime })
+        if (
+          e.name === 'KafkaJSAggregateError' &&
+          e.errors.some(error => error.type === 'UNKNOWN_TOPIC_OR_PARTITION')
+        ) {
+          logger.warn('Could not delete topics', {
+            error: e.message,
+            retryCount,
+            retryTime,
+            topics: e.errors.map(error => error.topic),
+          })
+          throw e
+        }
+
+        if (e.type === 'NOT_CONTROLLER') {
+          logger.warn('Could not delete topics', {
+            error: e.message,
+            retryCount,
+            retryTime,
+          })
           throw e
         }
 

src/errors.js

@@ -239,6 +239,14 @@ class KafkaJSCreateTopicError extends KafkaJSProtocolError {
   }
 }
 
+class KafkaJSDeleteTopicError extends KafkaJSProtocolError {
+  constructor(e, topicName) {
+    super(e)
+    this.topic = topicName
+    this.name = 'KafkaJSDeleteTopicError'
+  }
+}
+
 class KafkaJSAlterPartitionReassignmentsError extends KafkaJSProtocolError {
   constructor(e, topicName, partition) {
     super(e)
@@ -304,6 +312,7 @@ module.exports = {
   KafkaJSFetcherRebalanceError,
   KafkaJSNoBrokerAvailableError,
   KafkaJSAlterPartitionReassignmentsError,
+  KafkaJSDeleteTopicError,
   isRebalancing,
   isKafkaJSError,
 }

src/protocol/requests/deleteTopics/v0/response.js

@@ -1,5 +1,6 @@
 const Decoder = require('../../../decoder')
 const { failure, createErrorFromCode } = require('../../../error')
+const { KafkaJSDeleteTopicError, KafkaJSAggregateError } = require('../../../../errors')
 
 /**
  * DeleteTopics Response (Version: 0) => [topic_error_codes]
@@ -25,7 +26,12 @@ const decode = async rawData => {
 const parse = async data => {
   const topicsWithError = data.topicErrors.filter(({ errorCode }) => failure(errorCode))
   if (topicsWithError.length > 0) {
-    throw createErrorFromCode(topicsWithError[0].errorCode)
+    throw new KafkaJSAggregateError(
+      'Delete topics error',
+      topicsWithError.map(
+        error => new KafkaJSDeleteTopicError(createErrorFromCode(error.errorCode), error.topic)
+      )
+    )
   }
 
   return data