fix(node-integration-tests): Fix flaky kafkajs test race condition

dev-packages/node-integration-tests/suites/tracing/kafkajs/test.ts

@@ -1,3 +1,4 @@
+import type { TransactionEvent } from '@sentry/core';
 import { afterAll, describe, expect } from 'vitest';
 import { cleanupChildProcesses, createEsmAndCjsTests } from '../../../utils/runner';
 
@@ -8,16 +9,50 @@ describe('kafkajs', () => {
 
   createEsmAndCjsTests(__dirname, 'scenario.mjs', 'instrument.mjs', (createRunner, test) => {
     test('traces producers and consumers', { timeout: 60_000 }, async () => {
+      // The producer and consumer transactions can arrive in any order,
+      // so we collect them and assert after both have been received.
+      const receivedTransactions: TransactionEvent[] = [];
+
       await createRunner()
         .withDockerCompose({
           workingDirectory: [__dirname],
           readyMatches: ['9092'],
         })
         .expect({
-          transaction: {
-            transaction: 'send test-topic',
-            contexts: {
-              trace: expect.objectContaining({
+          transaction: (transaction: TransactionEvent) => {
+            receivedTransactions.push(transaction);
+          },
+        })
+        .expect({
+          transaction: (transaction: TransactionEvent) => {
+            receivedTransactions.push(transaction);
+
+            const producer = receivedTransactions.find(
+              t => t.contexts?.trace?.data?.['sentry.origin'] === 'auto.kafkajs.otel.producer',
+            );
+            const consumer = receivedTransactions.find(
+              t => t.contexts?.trace?.data?.['sentry.origin'] === 'auto.kafkajs.otel.consumer',
+            );
+
+            expect(producer).toBeDefined();
+            expect(consumer).toBeDefined();
+
+            for (const t of [producer, consumer]) {
+              // just to assert on the basic shape (for more straight-forward tests, this is usually done by the runner)
+              expect(t).toMatchObject({
+                event_id: expect.any(String),
+                timestamp: expect.anything(),
+                start_timestamp: expect.anything(),
+                spans: expect.any(Array),
+                type: 'transaction',
+              });
+            }
+
+            expect(producer!.transaction).toBe('send test-topic');
+            expect(consumer!.transaction).toBe('process test-topic');
+
+            expect(producer!.contexts?.trace).toMatchObject(
+              expect.objectContaining({
                 op: 'message',
                 status: 'ok',
                 data: expect.objectContaining({
@@ -28,14 +63,10 @@ describe('kafkajs', () => {
                   'sentry.origin': 'auto.kafkajs.otel.producer',
                 }),
               }),
-            },
-          },
-        })
-        .expect({
-          transaction: {
-            transaction: 'process test-topic',
-            contexts: {
-              trace: expect.objectContaining({
+            );
+
+            expect(consumer!.contexts?.trace).toMatchObject(
+              expect.objectContaining({
                 op: 'message',
                 status: 'ok',
                 data: expect.objectContaining({
@@ -46,7 +77,7 @@ describe('kafkajs', () => {
                   'sentry.origin': 'auto.kafkajs.otel.consumer',
                 }),
               }),
-            },
+            );
           },
         })
         .start()