feat(core): Add the dataCollection.queues option

MIGRATION.md

@@ -187,6 +187,7 @@ We've replaced `sendDefaultPii` with `dataCollection`, which controls each categ
 | `urlQueryParams`      | `true`                             | `true`               |
 | `genAI`               | inputs + outputs not collected     | inputs + outputs     |
 | `databaseQueryData`   | `false`                            | `true`               |
+| `queues`              | not collected                      | `true`               |
 | `stackFrameVariables` | `true`                             | `true`               |
 | `frameContextLines`   | `7`                                | `5`                  |
 
@@ -222,6 +223,7 @@ Sentry.init({
     urlQueryParams: { deny: ['forwarded', '-ip', 'remote-', 'via', '-user'] },
     genAI: { inputs: false, outputs: false },
     databaseQueryData: false,
+    queues: false,
     graphQL: { document: false, variables: false },
   },
 });

packages/core/src/types/datacollection.ts

@@ -86,6 +86,15 @@ export interface DataCollection {
    */
   databaseQueryData?: boolean;
 
+  /**
+   * Include arguments passed to tasks within queues.
+   *
+   * Structural metadata such as the messaging system, destination name, or operation is always
+   * collected.
+   * @default true
+   */
+  queues?: boolean;
+
   /**
    * Capture local variable values in stack frames.
    *

packages/core/src/utils/data-collection/resolveDataCollectionOptions.ts

@@ -9,6 +9,7 @@ const DEFAULTS: ResolvedDataCollection = {
   graphQL: { document: true, variables: true },
   genAI: { inputs: true, outputs: true },
   databaseQueryData: true,
+  queues: true,
   stackFrameVariables: true,
   frameContextLines: 5,
 };
@@ -41,6 +42,7 @@ export function resolveDataCollectionOptions(options: { dataCollection?: DataCol
       outputs: dc.genAI?.outputs ?? DEFAULTS.genAI.outputs,
     },
     databaseQueryData: dc.databaseQueryData ?? DEFAULTS.databaseQueryData,
+    queues: dc.queues ?? DEFAULTS.queues,
     stackFrameVariables: dc.stackFrameVariables ?? DEFAULTS.stackFrameVariables,
     frameContextLines: dc.frameContextLines ?? DEFAULTS.frameContextLines,
   };

packages/core/test/lib/utils/data-collection/resolveDataCollectionOptions.test.ts

@@ -11,6 +11,7 @@ describe('resolveDataCollectionOptions', () => {
     graphQL: { document: true, variables: true },
     genAI: { inputs: true, outputs: true },
     databaseQueryData: true,
+    queues: true,
     stackFrameVariables: true,
     frameContextLines: 5,
   };
@@ -120,6 +121,10 @@ describe('resolveDataCollectionOptions', () => {
       expect(result.databaseQueryData).toBe(false);
     });
 
+    it('supports turning off queue data', () => {
+      expect(resolveDataCollectionOptions({ dataCollection: { queues: false } }).queues).toBe(false);
+    });
+
     it('supports allow/deny list for stack frame variables', () => {
       expect(
         resolveDataCollectionOptions({ dataCollection: { stackFrameVariables: { allow: ['user'] } } })
@@ -147,7 +152,7 @@ describe('resolveDataCollectionOptions', () => {
     it('always returns all fields', () => {
       const result = resolveDataCollectionOptions({});
 
-      expect(Object.keys(result)).toHaveLength(10);
+      expect(Object.keys(result)).toHaveLength(11);
       expect(result).toHaveProperty('userInfo');
       expect(result).toHaveProperty('cookies');
       expect(result).toHaveProperty('httpHeaders');
@@ -162,6 +167,7 @@ describe('resolveDataCollectionOptions', () => {
       expect(result).toHaveProperty('genAI.inputs');
       expect(result).toHaveProperty('genAI.outputs');
       expect(result).toHaveProperty('databaseQueryData');
+      expect(result).toHaveProperty('queues');
       expect(result).toHaveProperty('stackFrameVariables');
       expect(result).toHaveProperty('frameContextLines');
     });

packages/server-utils/src/integrations/kafkajs/spans.ts

@@ -99,6 +99,17 @@ export function getLinksFromHeaders(headers: KafkaMessage['headers']): SpanLink[
   ];
 }
 
+/**
+ * The Kafka message key is producer-supplied payload data, so `dataCollection.queues` gates it.
+ * Everything else on the span (topic, partition, offset) is structural metadata and stays.
+ */
+function collectMessageKey(key: unknown, client = getClient()): string | undefined {
+  if (!key || client?.getDataCollectionOptions().queues === false) {
+    return undefined;
+  }
+  return String(key);
+}
+
 /** Starts an inactive consumer (process/receive) span carrying the kafkajs messaging attributes. */
 export function startConsumerSpan({ topic, message, operationType, links, attributes }: ConsumerSpanOptions): Span {
   // The batch "receive" span is named `poll`; per-message spans use the operation type verbatim.
@@ -119,7 +130,7 @@ export function startConsumerSpan({ topic, message, operationType, links, attrib
       [MESSAGING_DESTINATION_NAME]: topic,
       [MESSAGING_OPERATION_TYPE]: operationType,
       [MESSAGING_OPERATION_NAME]: operationName,
-      [ATTR_MESSAGING_KAFKA_MESSAGE_KEY]: message?.key ? String(message.key) : undefined,
+      [ATTR_MESSAGING_KAFKA_MESSAGE_KEY]: collectMessageKey(message?.key, client),
       [ATTR_MESSAGING_KAFKA_MESSAGE_TOMBSTONE]: message?.key && message.value === null ? true : undefined,
       [ATTR_MESSAGING_KAFKA_OFFSET]: message?.offset as string | undefined,
       // Mirror the upstream behavior of only tagging per-message processing spans (not the batch
@@ -138,7 +149,7 @@ export function startProducerSpan(topic: string, message: Message): Span {
       [SENTRY_KIND]: 'producer',
       [MESSAGING_SYSTEM]: MESSAGING_SYSTEM_VALUE_KAFKA,
       [MESSAGING_DESTINATION_NAME]: topic,
-      [ATTR_MESSAGING_KAFKA_MESSAGE_KEY]: message.key ? String(message.key) : undefined,
+      [ATTR_MESSAGING_KAFKA_MESSAGE_KEY]: collectMessageKey(message.key),
       [ATTR_MESSAGING_KAFKA_MESSAGE_TOMBSTONE]: message.key && message.value === null ? true : undefined,
       [ATTR_MESSAGING_DESTINATION_PARTITION_ID]:
         message.partition !== undefined ? String(message.partition) : undefined,