Handle request serialisation errors asynchronously (#154)

Author: baac0

src/network/connection.ts

@@ -350,7 +350,7 @@ export class Connection extends EventEmitter {
     } catch (err) {
       diagnostic.error = err as Error
       connectionsApiChannel.error.publish(diagnostic)
-      throw err
+      return callback(err, undefined as unknown as ReturnType)
     }
 
     writer.appendFrom(payload).prependLength()

test/clients/producer/producer.test.ts

@@ -4,6 +4,8 @@ import * as Prometheus from 'prom-client'
 import { kConnections } from '../../../src/clients/base/base.ts'
 import {
   type ClientDiagnosticEvent,
+  compressionsAlgorithms,
+  GenericError,
   initProducerIdV5,
   instancesChannel,
   MultipleErrors,
@@ -691,6 +693,30 @@ test('send should auto-initialize idempotent producer if needed', async t => {
   strictEqual(result.offsets?.length, 1)
 })
 
+test('send should handle synchronuous error during payload creation', async t => {
+  const producer = createProducer(t, { strict: true })
+  const testTopic = await createTopic(t)
+
+  const compression = 'lz4'
+  const expectedError = new GenericError('PLT_KFK_UNSUPPORTED_COMPRESSION', 'Avoid RUD')
+  t.mock.method(compressionsAlgorithms[compression], 'compressSync', () => { throw expectedError })
+
+  await rejects(
+
+    async () => {
+      await producer.send({
+        messages: [{ topic: testTopic, value: Buffer.from('auto-init-idempotent-message') }],
+        compression,
+      })
+    },
+    (error: any) => {
+      strictEqual(error instanceof AggregateError, true)
+      strictEqual(error.errors[0], expectedError)
+      return true
+    }
+  )
+})
+
 test('send should validate options in strict mode', async t => {
   const producer = createProducer(t, { strict: true })
   const testTopic = await createTopic(t)

test/network/connection.test.ts

@@ -752,17 +752,27 @@ test('Connection should handle request serialization errors', async t => {
     throw new Error('Parser error')
   }
 
-  // Send a request
-  await throws(() =>
+  const requestPromise = new Promise((resolve, reject) => {
     connection.send(
       0, // apiKey
       0, // apiVersion
       payloadFn as () => Writer,
       parser,
       false,
       false,
-      () => {}
-    ))
+      (err: any) => {
+        if (err) {
+          reject(err)
+        }
+
+        resolve(null)
+      }
+    )
+  })
+
+  await rejects(() => requestPromise, {
+    message: 'Serialization error'
+  })
 
   verifyTracingChannel()
 })