ref(node): Streamline kafkajs instrumentation

dev-packages/node-integration-tests/suites/tracing/kafkajs/scenario-error.mjs

@@ -0,0 +1,29 @@
+import { Kafka } from 'kafkajs';
+
+async function run() {
+  const kafka = new Kafka({
+    clientId: 'my-app',
+    brokers: ['localhost:9092'],
+    // The invalid-topic error is non-retriable, but disable retries anyway so the failing
+    // `send` rejects promptly and produces exactly one errored producer span.
+    retry: { retries: 0 },
+  });
+
+  const producer = kafka.producer();
+  await producer.connect();
+
+  // Topic names may not contain spaces, so the broker rejects this send. The producer span should
+  // be marked as errored.
+  await producer
+    .send({
+      topic: 'invalid topic name',
+      messages: [{ value: 'TEST_MESSAGE' }],
+    })
+    .catch(() => {
+      // swallow - we assert on the emitted span, not the thrown error
+    });
+
+  await producer.disconnect();
+}
+
+run();

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

@@ -83,4 +83,34 @@ describe('kafkajs', () => {
         .completed();
     });
   });
+
+  createEsmAndCjsTests(__dirname, 'scenario-error.mjs', 'instrument.mjs', (createRunner, test) => {
+    test('marks the producer span as errored when a send fails', { timeout: 90_000 }, async () => {
+      await createRunner()
+        .withDockerCompose({
+          workingDirectory: [__dirname],
+        })
+        .expect({
+          transaction: (transaction: TransactionEvent) => {
+            expect(transaction.transaction).toBe('send invalid topic name');
+            expect(transaction.contexts?.trace).toMatchObject(
+              expect.objectContaining({
+                op: 'message',
+                status: 'internal_error',
+                data: expect.objectContaining({
+                  'messaging.system': 'kafka',
+                  'messaging.destination.name': 'invalid topic name',
+                  'otel.kind': 'PRODUCER',
+                  'sentry.op': 'message',
+                  'sentry.origin': 'auto.kafkajs.otel.producer',
+                  'error.type': 'KafkaJSNonRetriableError',
+                }),
+              }),
+            );
+          },
+        })
+        .start()
+        .completed();
+    });
+  });
 });

packages/node/src/integrations/tracing/kafka/index.ts

@@ -1,22 +1,11 @@
 import { KafkaJsInstrumentation } from './vendored/instrumentation';
 import type { IntegrationFn } from '@sentry/core';
 import { defineIntegration } from '@sentry/core';
-import { addOriginToSpan, generateInstrumentOnce } from '@sentry/node-core';
+import { generateInstrumentOnce } from '@sentry/node-core';
 
 const INTEGRATION_NAME = 'Kafka';
 
-export const instrumentKafka = generateInstrumentOnce(
-  INTEGRATION_NAME,
-  () =>
-    new KafkaJsInstrumentation({
-      consumerHook(span) {
-        addOriginToSpan(span, 'auto.kafkajs.otel.consumer');
-      },
-      producerHook(span) {
-        addOriginToSpan(span, 'auto.kafkajs.otel.producer');
-      },
-    }),
-);
+export const instrumentKafka = generateInstrumentOnce(INTEGRATION_NAME, () => new KafkaJsInstrumentation());
 
 const _kafkaIntegration = (() => {
   return {