test: fix flaky confluenctinc kafka error (#7460)

BridgeAR · web-flow · commit 510563d09e17 · 2026-02-09T17:40:42.000Z
The broker needs some time after creating a new topic to settle in.

The old commented error test case could not work due to expecting
to catch a user error (which we should not). It is therefore
removed.
diff --git a/packages/datadog-plugin-confluentinc-kafka-javascript/test/index.spec.js b/packages/datadog-plugin-confluentinc-kafka-javascript/test/index.spec.js
@@ -65,6 +65,10 @@ describe('Plugin', () => {
               replicationFactor: 1,
             }],
           })
+
+          // `createTopics()` returns before leaders are guaranteed to be elected in this client.
+          // If we race ahead immediately, consumers/producers can stall on metadata/leader availability.
+          await waitForTopicReady(admin, testTopic)
         })
 
         afterEach(() => admin.disconnect())
@@ -153,17 +157,19 @@ describe('Plugin', () => {
                 type: 'worker',
               })
 
-              const consumerReceiveMessagePromise = new Promise(resolve => {
+              const consumerReceiveMessagePromise = /** @type {Promise<void>} */(new Promise((resolve, reject) => {
                 consumer.run({
                   eachMessage: () => {
                     resolve()
                   },
                 })
-              })
-              await sendMessages(kafka, testTopic, messages).then(
-                async () => await consumerReceiveMessagePromise
-              )
-              return expectedSpanPromise
+              }))
+
+              await Promise.all([
+                sendMessages(kafka, testTopic, messages),
+                consumerReceiveMessagePromise,
+                expectedSpanPromise,
+              ])
             })
 
             it('should run the consumer in the context of the consumer span', done => {
@@ -179,7 +185,7 @@ describe('Plugin', () => {
                 } catch (e) {
                   done(e)
                 } finally {
-                  eachMessage = () => {} // avoid being called for each message
+                  eachMessage = async () => {} // avoid being called for each message
                 }
               }
 
@@ -271,25 +277,25 @@ describe('Plugin', () => {
               dr_cb: true,
             })
 
-            await new Promise((resolve, reject) => {
+            await /** @type {Promise<void>} */(new Promise((resolve, reject) => {
               nativeProducer.connect({}, (err) => {
                 if (err) {
                   return reject(err)
                 }
                 resolve()
               })
-            })
+            }))
           })
 
           afterEach(async () => {
-            await new Promise((resolve, reject) => {
+            await /** @type {Promise<void>} */(new Promise((resolve, reject) => {
               nativeProducer.disconnect((err) => {
                 if (err) {
                   return reject(err)
                 }
                 resolve()
               })
-            })
+            }))
           })
 
           describe('producer', () => {
@@ -354,30 +360,30 @@ describe('Plugin', () => {
                 'auto.offset.reset': 'earliest',
               })
 
-              await new Promise((resolve, reject) => {
+              await /** @type {Promise<void>} */(new Promise((resolve, reject) => {
                 nativeConsumer.connect({}, (err) => {
                   if (err) {
                     return reject(err)
                   }
                   resolve()
                 })
-              })
+              }))
             })
 
             afterEach(async () => {
               await nativeConsumer.unsubscribe()
-              await new Promise((resolve, reject) => {
+              await /** @type {Promise<void>} */(new Promise((resolve, reject) => {
                 nativeConsumer.disconnect((err) => {
                   if (err) {
                     return reject(err)
                   }
                   resolve()
                 })
-              })
+              }))
             })
 
             function consume (consumer, producer, topic, message, timeoutMs = 9500) {
-              return new Promise((resolve, reject) => {
+              return /** @type {Promise<void>} */(new Promise((resolve, reject) => {
                 const timeoutId = setTimeout(() => {
                   reject(new Error(`Timeout: Did not consume message on topic "${topic}" within ${timeoutMs}ms`))
                 }, timeoutMs)
@@ -408,7 +414,7 @@ describe('Plugin', () => {
                 }
                 doConsume()
                 producer.produce(topic, null, message, 'native-consumer-key')
-              })
+              }))
             }
 
             it('should be instrumented', async () => {
@@ -458,33 +464,6 @@ describe('Plugin', () => {
 
               return expectedSpanPromise
             })
-
-            // TODO: Fix this test case, fails with 'done() called multiple times'
-            // it('should be instrumented with error', async () => {
-            //   const fakeError = new Error('Oh No!')
-
-            //   const expectedSpanPromise = agent.assertSomeTraces(traces => {
-            //     const errorSpans = traces[0].filter(span => span.error === 1)
-            //     expect(errorSpans.length).to.be.at.least(1)
-
-            //     const errorSpan = errorSpans[0]
-            //     expect(errorSpan).to.exist
-            //     expect(errorSpan.name).to.equal(expectedSchema.receive.opName)
-            //     expect(errorSpan.meta).to.include({
-            //       component: 'confluentinc-kafka-javascript'
-            //     })
-
-            //     expect(errorSpan.meta[ERROR_TYPE]).to.equal(fakeError.name)
-            //     expect(errorSpan.meta[ERROR_MESSAGE]).to.equal(fakeError.message)
-            //   })
-
-            //   nativeConsumer.consume(1, (err, messages) => {
-            //     // Ensure we resolve before throwing
-            //     throw fakeError
-            //   })
-
-            //   return expectedSpanPromise
-            // })
           })
         })
       })
@@ -511,3 +490,28 @@ async function sendMessages (kafka, topic, messages) {
   })
   await producer.disconnect()
 }
+
+async function waitForTopicReady (admin, topic, timeoutMs = 20000) {
+  if (typeof admin?.fetchTopicMetadata !== 'function') return
+
+  const start = Date.now()
+  while ((Date.now() - start) < timeoutMs) {
+    try {
+      const meta = await admin.fetchTopicMetadata({ topics: [topic], timeout: 1000 })
+      const topicMeta = Array.isArray(meta) ? meta[0] : meta?.topics?.[0]
+
+      const partitions = topicMeta?.partitions
+      if (Array.isArray(partitions) &&
+          partitions.length > 0 &&
+          partitions.every(p => typeof p.leader === 'number' && p.leader >= 0)) {
+        return
+      }
+    } catch {
+      // Topic creation is async; metadata/leader errors can be transient.
+    }
+
+    await new Promise(resolve => setTimeout(resolve, 50))
+  }
+
+  throw new Error(`Timeout: Topic "${topic}" metadata was not ready within ${timeoutMs}ms`)
+}
