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

The kafkajs integration test asserted producer and consumer transactions
in a fixed order, but they can arrive in either order due to Kafka's
async nature.

To fix the flake, we collect both transactions via callbacks, then
assert after both have arrived using `find()` by transaction name
instead of relying on arrival order

closes #20121

---------

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: Lukas Stracke <lukas.stracke@sentry.io>

Author: Copilot

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()