feat(core): Queue Instrumentation for Kafka

.cursor/rules/overview_dev.mdc

@@ -66,6 +66,15 @@ Use the `fetch_rules` tool to include these rules when working on specific areas
   - `SentryMetricsEvent`, `SentryMetricsEvents`
   - `SentryOptions.getMetrics()`, `beforeSend` callback
 
+- **`queues`**: Use when working with:
+  - Sentry Queues product data or messaging span conventions
+  - Queue tracing spans/transactions (`queue.publish`, `queue.process`)
+  - `enableQueueTracing` option and `sentry.enable-queue-tracing`
+  - Kafka instrumentation (`sentry-kafka`, `SentryKafkaProducer`, `SentryKafkaConsumerTracing`)
+  - Spring Kafka queue auto-instrumentation and `SentryKafkaRecordInterceptor`
+  - Messaging span data (`messaging.system`, `messaging.destination.name`, receive latency, retry count)
+  - `sentry-task-enqueued-time` header and distributed trace propagation through queues
+
 - **`continuous_profiling_jvm`**: Use when working with:
   - JVM continuous profiling (`sentry-async-profiler` module)
   - `IContinuousProfiler`, `JavaContinuousProfiler`
@@ -118,6 +127,7 @@ Use the `fetch_rules` tool to include these rules when working on specific areas
    - System test/e2e/sample → `e2e_tests`
    - Feature flag/addFeatureFlag/flag evaluation → `feature_flags`
    - Metrics/count/distribution/gauge → `metrics`
+   - Queues/queue tracing/Kafka/Spring Kafka/queue.publish/queue.process/enableQueueTracing/messaging spans → `queues`
    - PR/pull request/stacked PR/stack → `pr`
    - JVM continuous profiling/async-profiler/JFR/ProfileChunk → `continuous_profiling_jvm`
    - Android continuous profiling/AndroidProfiler/frame metrics/method tracing → no dedicated rule yet; inspect the code directly

.cursor/rules/pr.mdc

@@ -258,3 +258,5 @@ git push
 **Never merge into the collection branch.** Syncing only happens between stack PR branches. The collection branch is untouched until the user merges PRs through GitHub.
 
 Prefer merge over rebase — it preserves commit history, doesn't invalidate existing review comments, and avoids the need for force-pushing. Only rebase if explicitly requested.
+
+**Never amend or force-push stack branches.** Do not use `git commit --amend`, `--force`, or `--force-with-lease` on branches that are part of a stack. Amending a pushed commit requires a force-push, which can cause GitHub to auto-merge or auto-close other PRs in the stack. If a commit needs fixing, add a new fixup commit instead.

.cursor/rules/queues.mdc

@@ -0,0 +1,82 @@
+---
+alwaysApply: false
+description: Sentry Queues module and Java SDK queue tracing
+---
+# Sentry Queues and Java SDK Queue Tracing
+
+## Product model
+
+Sentry Queues is built from tracing data. SDKs mark queue work with queue-specific span operations and messaging span data so Sentry can identify producers, consumers, destinations, latency, and failures.
+
+The important concepts are:
+- `queue.publish`: a span for enqueueing/publishing a message to a queue or topic.
+- `queue.process`: a transaction for processing a dequeued message.
+- Messaging span data, especially:
+  - `messaging.system` (for example `kafka`)
+  - `messaging.destination.name` (queue/topic name)
+  - `messaging.message.id`
+  - `messaging.message.retry.count`
+  - `messaging.message.body.size`
+  - `messaging.message.envelope.size`
+  - `messaging.message.receive.latency`
+- Distributed tracing headers (`sentry-trace` and `baggage`) link producer-side work to consumer-side processing.
+- Queue receive latency is the time a message spent waiting between publish/enqueue and processing. For Java Kafka, this comes from the `sentry-task-enqueued-time` header that the producer writes and the consumer reads.
+
+The Queues UI is not backed by a separate Java event type. The Java SDK contributes data through spans/transactions with the expected operations, trace context, statuses, and messaging attributes.
+
+## Java SDK implementation
+
+Queue tracing is opt-in. `SentryOptions.isEnableQueueTracing()` defaults to `false` and can be enabled with `setEnableQueueTracing(true)` or external config key `enable-queue-tracing` (`sentry.enable-queue-tracing` in Spring Boot). Captured queue spans/transactions still depend on tracing being enabled and sampled.
+
+Kafka support lives in `sentry-kafka`:
+- `SentryKafkaProducer.wrap(Producer)` wraps Kafka `Producer.send(...)` calls.
+  - Creates a `queue.publish` child span when there is an active span.
+  - Sets `messaging.system=kafka` and `messaging.destination.name=<topic>`.
+  - Injects `sentry-trace`, `baggage`, and `sentry-task-enqueued-time` headers.
+  - Still injects tracing/enqueued-time headers when queue tracing is enabled but there is no active span, so background producers can link to consumers.
+  - Finishes the span from the Kafka callback with `OK` or `INTERNAL_ERROR`.
+- `SentryKafkaConsumerTracing.withTracing(record, callback)` is the manual raw-Kafka consumer helper.
+  - Forks root scopes for the processing lifecycle and makes them current.
+  - Continues the trace from Kafka headers.
+  - Starts a `queue.process` transaction bound to scope when tracing is enabled.
+  - Sets Kafka messaging data, body size, retry count, and receive latency when available.
+  - Finishes with `OK` or `INTERNAL_ERROR` and never lets instrumentation failures break customer processing.
+
+Spring Kafka support lives in `sentry-spring`, `sentry-spring-jakarta`, and `sentry-spring-7`:
+- `SentryKafkaProducerBeanPostProcessor` installs a producer post-processor on `DefaultKafkaProducerFactory` and wraps created producers with `SentryKafkaProducer.wrap(...)`.
+- `SentryKafkaConsumerBeanPostProcessor` installs `SentryKafkaRecordInterceptor` on listener container factories.
+- `SentryKafkaRecordInterceptor` starts/finishes `queue.process` transactions around listener processing, continues traces from headers, forks scopes for the record lifecycle, and preserves any existing delegate interceptor.
+- Spring Boot auto-configuration registers both post-processors only when Spring Kafka and `sentry-kafka` are present and `sentry.enable-queue-tracing=true`.
+- Spring Boot queue auto-configuration is disabled when Sentry OpenTelemetry integration classes are present to avoid duplicate Kafka instrumentation.
+
+## Trace origins and suppression
+
+Queue instrumentation sets span origins so it can be identified and suppressed with `ignoredSpanOrigins`:
+- Raw Kafka producer: `auto.queue.kafka.producer`
+- Raw Kafka consumer helper: `manual.queue.kafka.consumer`
+- Spring Kafka producer: `auto.queue.spring.kafka.producer`, `auto.queue.spring_jakarta.kafka.producer`, `auto.queue.spring7.kafka.producer`
+- Spring Kafka consumer: `auto.queue.spring.kafka.consumer`, `auto.queue.spring_jakarta.kafka.consumer`, `auto.queue.spring7.kafka.consumer`
+
+## Files to inspect when changing queue tracing
+
+- Core option and conventions:
+  - `sentry/src/main/java/io/sentry/SentryOptions.java`
+  - `sentry/src/main/java/io/sentry/ExternalOptions.java`
+  - `sentry/src/main/java/io/sentry/SpanDataConvention.java`
+- Raw Kafka:
+  - `sentry-kafka/src/main/java/io/sentry/kafka/SentryKafkaProducer.java`
+  - `sentry-kafka/src/main/java/io/sentry/kafka/SentryKafkaConsumerTracing.java`
+  - `sentry-kafka/src/test/kotlin/io/sentry/kafka/*Test.kt`
+- Spring Kafka:
+  - `sentry-spring*/src/main/java/io/sentry/**/kafka/*`
+  - `sentry-spring*/src/test/kotlin/io/sentry/**/kafka/*Test.kt`
+  - `sentry-spring-boot*/src/main/java/io/sentry/**/SentryAutoConfiguration.java`
+  - `sentry-spring-boot*/src/test/kotlin/io/sentry/**/SentryKafkaAutoConfigurationTest.kt`
+
+## Related rules
+
+Also fetch:
+- `options` when changing `enableQueueTracing` or configuration surfaces.
+- `scopes` when changing consumer scope forking/lifecycle.
+- `opentelemetry` when changing coexistence with OTel auto-instrumentation.
+- `api` when changing public Kafka APIs or option methods.

CHANGELOG.md

@@ -19,6 +19,16 @@
     .configurator { it.isUseShakeGesture = true }
     .create()
   ```
+- Add support for Kafka ([#5249](https://github.com/getsentry/sentry-java/pull/5249))
+  - You will need to add the `sentry-kafka` dependency and opt-in via the new option.
+    - Set `options.setEnableQueueTracing(true)` on `Sentry.init`
+    - Or set `sentry.enable-queue-tracing=true` in `application.properties`
+  - For Spring Boot Kafka is auto instrumented and no further configuration is needed.
+    - also see https://docs.sentry.io/platforms/java/guides/spring-boot/integrations/kafka/
+  - When using `kafka-clients` directly
+    - you need to wrap your `KafkaProducer` via `SentryKafkaProducer.wrap(kafkaProducer)` to get `queue.publish` spans
+    - and you may use our `SentryKafkaConsumerTracing.withTracing` helper to instrument the consumer side manually.
+    - also see https://docs.sentry.io/platforms/java/integrations/kafka/
 
 ### Fixes
 

README.md

@@ -35,6 +35,7 @@ Sentry SDK for Java and Android
 | sentry                                  | [![Maven Central Version](https://img.shields.io/maven-central/v/io.sentry/sentry?style=for-the-badge&logo=sentry&color=green)](https://central.sonatype.com/artifact/io.sentry/sentry)                                  | 21 |
 | sentry-jul                              | [![Maven Central Version](https://img.shields.io/maven-central/v/io.sentry/sentry-jul?style=for-the-badge&logo=sentry&color=green)](https://central.sonatype.com/artifact/io.sentry/sentry-jul)                              |
 | sentry-jdbc                             | [![Maven Central Version](https://img.shields.io/maven-central/v/io.sentry/sentry-jdbc?style=for-the-badge&logo=sentry&color=green)](https://central.sonatype.com/artifact/io.sentry/sentry-jdbc)                             |
+| sentry-kafka                            | [![Maven Central Version](https://img.shields.io/maven-central/v/io.sentry/sentry-kafka?style=for-the-badge&logo=sentry&color=green)](https://central.sonatype.com/artifact/io.sentry/sentry-kafka)                            |
 | sentry-apollo                           | [![Maven Central Version](https://img.shields.io/maven-central/v/io.sentry/sentry-apollo?style=for-the-badge&logo=sentry&color=green)](https://central.sonatype.com/artifact/io.sentry/sentry-apollo)                           | 21 |
 | sentry-apollo-3                         | [![Maven Central Version](https://img.shields.io/maven-central/v/io.sentry/sentry-apollo-3?style=for-the-badge&logo=sentry&color=green)](https://central.sonatype.com/artifact/io.sentry/sentry-apollo-3)                         | 21 |
 | sentry-apollo-4                         | [![Maven Central Version](https://img.shields.io/maven-central/v/io.sentry/sentry-apollo-4?style=for-the-badge&logo=sentry&color=green)](https://central.sonatype.com/artifact/io.sentry/sentry-apollo-4)                         | 21 |

buildSrc/src/main/java/Config.kt

@@ -80,6 +80,7 @@ object Config {
         val SENTRY_JCACHE_SDK_NAME = "$SENTRY_JAVA_SDK_NAME.jcache"
         val SENTRY_QUARTZ_SDK_NAME = "$SENTRY_JAVA_SDK_NAME.quartz"
         val SENTRY_JDBC_SDK_NAME = "$SENTRY_JAVA_SDK_NAME.jdbc"
+        val SENTRY_KAFKA_SDK_NAME = "$SENTRY_JAVA_SDK_NAME.kafka"
         val SENTRY_OPENFEATURE_SDK_NAME = "$SENTRY_JAVA_SDK_NAME.openfeature"
         val SENTRY_LAUNCHDARKLY_SERVER_SDK_NAME = "$SENTRY_JAVA_SDK_NAME.launchdarkly-server"
         val SENTRY_LAUNCHDARKLY_ANDROID_SDK_NAME = "$SENTRY_ANDROID_SDK_NAME.launchdarkly"

gradle/libs.versions.toml

@@ -184,6 +184,10 @@ springboot3-starter-security = { module = "org.springframework.boot:spring-boot-
 springboot3-starter-jdbc = { module = "org.springframework.boot:spring-boot-starter-jdbc", version.ref = "springboot3" }
 springboot3-starter-actuator = { module = "org.springframework.boot:spring-boot-starter-actuator", version.ref = "springboot3" }
 springboot3-starter-cache = { module = "org.springframework.boot:spring-boot-starter-cache", version.ref = "springboot3" }
+spring-kafka2 = { module = "org.springframework.kafka:spring-kafka", version = "2.8.11" }
+spring-kafka3 = { module = "org.springframework.kafka:spring-kafka", version = "3.3.5" }
+spring-kafka4 = { module = "org.springframework.kafka:spring-kafka" }
+kafka-clients = { module = "org.apache.kafka:kafka-clients", version = "3.8.1" }
 springboot4-otel = { module = "io.opentelemetry.instrumentation:opentelemetry-spring-boot-starter", version.ref = "otelInstrumentation" }
 springboot4-resttestclient = { module = "org.springframework.boot:spring-boot-resttestclient", version.ref = "springboot4" }
 springboot4-starter = { module = "org.springframework.boot:spring-boot-starter", version.ref = "springboot4" }
@@ -200,6 +204,7 @@ springboot4-starter-webclient = { module = "org.springframework.boot:spring-boot
 springboot4-starter-jdbc = { module = "org.springframework.boot:spring-boot-starter-jdbc", version.ref = "springboot4" }
 springboot4-starter-actuator = { module = "org.springframework.boot:spring-boot-starter-actuator", version.ref = "springboot4" }
 springboot4-starter-cache = { module = "org.springframework.boot:spring-boot-starter-cache", version.ref = "springboot4" }
+springboot4-starter-kafka = { module = "org.springframework.boot:spring-boot-starter-kafka", version.ref = "springboot4" }
 timber = { module = "com.jakewharton.timber:timber", version = "4.7.1" }
 
 # Animalsniffer signature

sentry-kafka/README.md

@@ -0,0 +1,5 @@
+# sentry-kafka
+
+This module provides Kafka-native queue instrumentation for applications using `kafka-clients` directly.
+
+Spring users should use the Sentry Spring (Boot) SDKs, which provide higher-fidelity consumer instrumentation via Spring Kafka hooks.

sentry-kafka/api/sentry-kafka.api

@@ -0,0 +1,19 @@
+public final class io/sentry/kafka/BuildConfig {
+	public static final field SENTRY_KAFKA_SDK_NAME Ljava/lang/String;
+	public static final field VERSION_NAME Ljava/lang/String;
+}
+
+public final class io/sentry/kafka/SentryKafkaConsumerTracing {
+	public static final field TRACE_ORIGIN Ljava/lang/String;
+	public static fun withTracing (Lorg/apache/kafka/clients/consumer/ConsumerRecord;Ljava/lang/Runnable;)V
+	public static fun withTracing (Lorg/apache/kafka/clients/consumer/ConsumerRecord;Ljava/util/concurrent/Callable;)Ljava/lang/Object;
+}
+
+public final class io/sentry/kafka/SentryKafkaProducer {
+	public static final field SENTRY_ENQUEUED_TIME_HEADER Ljava/lang/String;
+	public static final field TRACE_ORIGIN Ljava/lang/String;
+	public static fun wrap (Lorg/apache/kafka/clients/producer/Producer;)Lorg/apache/kafka/clients/producer/Producer;
+	public static fun wrap (Lorg/apache/kafka/clients/producer/Producer;Lio/sentry/IScopes;)Lorg/apache/kafka/clients/producer/Producer;
+	public static fun wrap (Lorg/apache/kafka/clients/producer/Producer;Lio/sentry/IScopes;Ljava/lang/String;)Lorg/apache/kafka/clients/producer/Producer;
+}
+

sentry-kafka/build.gradle.kts

@@ -0,0 +1,83 @@
+import net.ltgt.gradle.errorprone.errorprone
+import org.jetbrains.kotlin.gradle.tasks.KotlinCompile
+
+plugins {
+  `java-library`
+  id("io.sentry.javadoc")
+  alias(libs.plugins.kotlin.jvm)
+  jacoco
+  alias(libs.plugins.errorprone)
+  alias(libs.plugins.gradle.versions)
+  alias(libs.plugins.buildconfig)
+}
+
+tasks.withType<KotlinCompile>().configureEach {
+  compilerOptions.jvmTarget = org.jetbrains.kotlin.gradle.dsl.JvmTarget.JVM_1_8
+}
+
+dependencies {
+  api(projects.sentry)
+  compileOnly(libs.kafka.clients)
+  compileOnly(libs.jetbrains.annotations)
+  compileOnly(libs.nopen.annotations)
+
+  errorprone(libs.errorprone.core)
+  errorprone(libs.nopen.checker)
+  errorprone(libs.nullaway)
+
+  // tests
+  testImplementation(projects.sentryTestSupport)
+  testImplementation(kotlin(Config.kotlinStdLib))
+  testImplementation(libs.kotlin.test.junit)
+  testImplementation(libs.mockito.kotlin)
+  testImplementation(libs.mockito.inline)
+  testImplementation(libs.kafka.clients)
+}
+
+configure<SourceSetContainer> { test { java.srcDir("src/test/java") } }
+
+jacoco { toolVersion = libs.versions.jacoco.get() }
+
+tasks.jacocoTestReport {
+  reports {
+    xml.required.set(true)
+    html.required.set(false)
+  }
+}
+
+tasks {
+  jacocoTestCoverageVerification {
+    violationRules { rule { limit { minimum = Config.QualityPlugins.Jacoco.minimumCoverage } } }
+  }
+  check {
+    dependsOn(jacocoTestCoverageVerification)
+    dependsOn(jacocoTestReport)
+  }
+}
+
+tasks.withType<JavaCompile>().configureEach {
+  options.errorprone {
+    check("NullAway", net.ltgt.gradle.errorprone.CheckSeverity.ERROR)
+    option("NullAway:AnnotatedPackages", "io.sentry")
+  }
+}
+
+buildConfig {
+  useJavaOutput()
+  packageName("io.sentry.kafka")
+  buildConfigField("String", "SENTRY_KAFKA_SDK_NAME", "\"${Config.Sentry.SENTRY_KAFKA_SDK_NAME}\"")
+  buildConfigField("String", "VERSION_NAME", "\"${project.version}\"")
+}
+
+tasks.jar {
+  manifest {
+    attributes(
+      "Sentry-Version-Name" to project.version,
+      "Sentry-SDK-Name" to Config.Sentry.SENTRY_KAFKA_SDK_NAME,
+      "Sentry-SDK-Package-Name" to "maven:io.sentry:sentry-kafka",
+      "Implementation-Vendor" to "Sentry",
+      "Implementation-Title" to project.name,
+      "Implementation-Version" to project.version,
+    )
+  }
+}