feat(spring): [Queue Instrumentation 40] Add Spring Boot 4 Kafka tracing

CHANGELOG.md

@@ -4,6 +4,7 @@
 
 ### Features
 
+- Add Kafka queue tracing for Spring Boot 4 ([#5348](https://github.com/getsentry/sentry-java/pull/5348))
 - Add `sentry-kafka` module for Kafka queue instrumentation without Spring ([#5288](https://github.com/getsentry/sentry-java/pull/5288))
 - Add Kafka queue tracing for Spring Boot 3 ([#5254](https://github.com/getsentry/sentry-java/pull/5254)), ([#5255](https://github.com/getsentry/sentry-java/pull/5255)), ([#5256](https://github.com/getsentry/sentry-java/pull/5256))
 - Add `enableQueueTracing` option and messaging span data conventions ([#5250](https://github.com/getsentry/sentry-java/pull/5250))

gradle/libs.versions.toml

@@ -184,6 +184,7 @@ springboot3-starter-jdbc = { module = "org.springframework.boot:spring-boot-star
 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-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" }

sentry-samples/sentry-samples-spring-boot-4-opentelemetry-noagent/build.gradle.kts

@@ -58,6 +58,10 @@ dependencies {
   implementation(projects.sentryOpentelemetry.sentryOpentelemetryAgentlessSpring)
   implementation(projects.sentryAsyncProfiler)
 
+  // kafka
+  implementation(libs.spring.kafka4)
+  implementation(projects.sentryKafka)
+
   // database query tracing
   implementation(projects.sentryJdbc)
   runtimeOnly(libs.hsqldb)

sentry-samples/sentry-samples-spring-boot-4-opentelemetry-noagent/src/main/java/io/sentry/samples/spring/boot4/queues/kafka/KafkaConsumer.java

@@ -0,0 +1,19 @@
+package io.sentry.samples.spring.boot4.queues.kafka;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.context.annotation.Profile;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Component;
+
+@Component
+@Profile("kafka")
+public class KafkaConsumer {
+
+  private static final Logger logger = LoggerFactory.getLogger(KafkaConsumer.class);
+
+  @KafkaListener(topics = "sentry-topic", groupId = "sentry-sample-group")
+  public void listen(String message) {
+    logger.info("Received message: {}", message);
+  }
+}

sentry-samples/sentry-samples-spring-boot-4-opentelemetry-noagent/src/main/java/io/sentry/samples/spring/boot4/queues/kafka/KafkaController.java

@@ -0,0 +1,26 @@
+package io.sentry.samples.spring.boot4.queues.kafka;
+
+import org.springframework.context.annotation.Profile;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+@RestController
+@Profile("kafka")
+@RequestMapping("/kafka")
+public class KafkaController {
+
+  private final KafkaTemplate<String, String> kafkaTemplate;
+
+  public KafkaController(KafkaTemplate<String, String> kafkaTemplate) {
+    this.kafkaTemplate = kafkaTemplate;
+  }
+
+  @GetMapping("/produce")
+  String produce(@RequestParam(defaultValue = "hello from sentry!") String message) {
+    kafkaTemplate.send("sentry-topic", message);
+    return "Message sent: " + message;
+  }
+}

sentry-samples/sentry-samples-spring-boot-4-opentelemetry-noagent/src/main/resources/application-kafka.properties

@@ -0,0 +1,12 @@
+# Kafka — activate with: --spring.profiles.active=kafka
+sentry.enable-queue-tracing=true
+
+spring.kafka.bootstrap-servers=localhost:9092
+spring.kafka.consumer.group-id=sentry-sample-group
+spring.kafka.consumer.auto-offset-reset=earliest
+spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
+spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
+spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
+spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
+
+logging.level.org.apache.kafka=warn

sentry-samples/sentry-samples-spring-boot-4-opentelemetry-noagent/src/test/kotlin/io/sentry/systemtest/KafkaOtelCoexistenceSystemTest.kt

@@ -0,0 +1,41 @@
+package io.sentry.systemtest
+
+import io.sentry.systemtest.util.TestHelper
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import org.junit.Before
+
+class KafkaOtelCoexistenceSystemTest {
+  lateinit var testHelper: TestHelper
+
+  @Before
+  fun setup() {
+    testHelper = TestHelper("http://localhost:8080")
+    testHelper.reset()
+  }
+
+  @Test
+  fun `Sentry Kafka integration is suppressed when OTel is active`() {
+    val restClient = testHelper.restClient
+
+    restClient.produceKafkaMessage("otel-coexistence-test")
+    assertEquals(200, restClient.lastKnownStatusCode)
+
+    testHelper.ensureTransactionReceived { transaction, _ ->
+      transaction.transaction == "GET /kafka/produce" &&
+        transaction.sdk?.integrationSet?.contains("SpringKafka") != true &&
+        transaction.spans.any { span ->
+          span.op == "queue.publish" &&
+            span.origin == "auto.opentelemetry" &&
+            span.data?.get("messaging.system") == "kafka"
+        }
+    }
+
+    testHelper.ensureTransactionReceived { transaction, _ ->
+      transaction.contexts.trace?.operation == "queue.process" &&
+        transaction.contexts.trace?.origin == "auto.opentelemetry" &&
+        transaction.contexts.trace?.data?.get("messaging.system") == "kafka" &&
+        transaction.sdk?.integrationSet?.contains("SpringKafka") != true
+    }
+  }
+}

sentry-samples/sentry-samples-spring-boot-4-opentelemetry/build.gradle.kts

@@ -59,6 +59,10 @@ dependencies {
   implementation(projects.sentryAsyncProfiler)
   implementation(libs.otel)
 
+  // kafka
+  implementation(libs.spring.kafka4)
+  implementation(projects.sentryKafka)
+
   // cache tracing
   implementation(libs.springboot4.starter.cache)
   implementation(libs.caffeine)

sentry-samples/sentry-samples-spring-boot-4-opentelemetry/src/main/java/io/sentry/samples/spring/boot4/queues/kafka/KafkaConsumer.java

@@ -0,0 +1,19 @@
+package io.sentry.samples.spring.boot4.queues.kafka;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.context.annotation.Profile;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Component;
+
+@Component
+@Profile("kafka")
+public class KafkaConsumer {
+
+  private static final Logger logger = LoggerFactory.getLogger(KafkaConsumer.class);
+
+  @KafkaListener(topics = "sentry-topic", groupId = "sentry-sample-group")
+  public void listen(String message) {
+    logger.info("Received message: {}", message);
+  }
+}

sentry-samples/sentry-samples-spring-boot-4-opentelemetry/src/main/java/io/sentry/samples/spring/boot4/queues/kafka/KafkaController.java

@@ -0,0 +1,26 @@
+package io.sentry.samples.spring.boot4.queues.kafka;
+
+import org.springframework.context.annotation.Profile;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+@RestController
+@Profile("kafka")
+@RequestMapping("/kafka")
+public class KafkaController {
+
+  private final KafkaTemplate<String, String> kafkaTemplate;
+
+  public KafkaController(KafkaTemplate<String, String> kafkaTemplate) {
+    this.kafkaTemplate = kafkaTemplate;
+  }
+
+  @GetMapping("/produce")
+  String produce(@RequestParam(defaultValue = "hello from sentry!") String message) {
+    kafkaTemplate.send("sentry-topic", message);
+    return "Message sent: " + message;
+  }
+}

sentry-samples/sentry-samples-spring-boot-4-opentelemetry/src/main/resources/application-kafka.properties

@@ -0,0 +1,12 @@
+# Kafka — activate with: --spring.profiles.active=kafka
+sentry.enable-queue-tracing=true
+
+spring.kafka.bootstrap-servers=localhost:9092
+spring.kafka.consumer.group-id=sentry-sample-group
+spring.kafka.consumer.auto-offset-reset=earliest
+spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
+spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
+spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
+spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
+
+logging.level.org.apache.kafka=warn

sentry-samples/sentry-samples-spring-boot-4-opentelemetry/src/test/kotlin/io/sentry/systemtest/KafkaOtelCoexistenceSystemTest.kt

@@ -0,0 +1,41 @@
+package io.sentry.systemtest
+
+import io.sentry.systemtest.util.TestHelper
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import org.junit.Before
+
+class KafkaOtelCoexistenceSystemTest {
+  lateinit var testHelper: TestHelper
+
+  @Before
+  fun setup() {
+    testHelper = TestHelper("http://localhost:8080")
+    testHelper.reset()
+  }
+
+  @Test
+  fun `Sentry Kafka integration is suppressed when OTel is active`() {
+    val restClient = testHelper.restClient
+
+    restClient.produceKafkaMessage("otel-coexistence-test")
+    assertEquals(200, restClient.lastKnownStatusCode)
+
+    testHelper.ensureTransactionReceived { transaction, _ ->
+      transaction.transaction == "GET /kafka/produce" &&
+        transaction.sdk?.integrationSet?.contains("SpringKafka") != true &&
+        transaction.spans.any { span ->
+          span.op == "queue.publish" &&
+            span.origin == "auto.opentelemetry" &&
+            span.data?.get("messaging.system") == "kafka"
+        }
+    }
+
+    testHelper.ensureTransactionReceived { transaction, _ ->
+      transaction.contexts.trace?.operation == "queue.process" &&
+        transaction.contexts.trace?.origin == "auto.opentelemetry" &&
+        transaction.contexts.trace?.data?.get("messaging.system") == "kafka" &&
+        transaction.sdk?.integrationSet?.contains("SpringKafka") != true
+    }
+  }
+}

sentry-samples/sentry-samples-spring-boot-4/build.gradle.kts

@@ -61,6 +61,10 @@ dependencies {
   implementation(libs.springboot4.starter.cache)
   implementation(libs.caffeine)
 
+  // kafka
+  implementation(libs.spring.kafka4)
+  implementation(projects.sentryKafka)
+
   // database query tracing
   implementation(projects.sentryJdbc)
   runtimeOnly(libs.hsqldb)

sentry-samples/sentry-samples-spring-boot-4/src/main/java/io/sentry/samples/spring/boot4/queues/kafka/KafkaConsumer.java

@@ -0,0 +1,19 @@
+package io.sentry.samples.spring.boot4.queues.kafka;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.context.annotation.Profile;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Component;
+
+@Component
+@Profile("kafka")
+public class KafkaConsumer {
+
+  private static final Logger logger = LoggerFactory.getLogger(KafkaConsumer.class);
+
+  @KafkaListener(topics = "sentry-topic", groupId = "sentry-sample-group")
+  public void listen(String message) {
+    logger.info("Received message: {}", message);
+  }
+}

sentry-samples/sentry-samples-spring-boot-4/src/main/java/io/sentry/samples/spring/boot4/queues/kafka/KafkaController.java

@@ -0,0 +1,26 @@
+package io.sentry.samples.spring.boot4.queues.kafka;
+
+import org.springframework.context.annotation.Profile;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+@RestController
+@Profile("kafka")
+@RequestMapping("/kafka")
+public class KafkaController {
+
+  private final KafkaTemplate<String, String> kafkaTemplate;
+
+  public KafkaController(KafkaTemplate<String, String> kafkaTemplate) {
+    this.kafkaTemplate = kafkaTemplate;
+  }
+
+  @GetMapping("/produce")
+  String produce(@RequestParam(defaultValue = "hello from sentry!") String message) {
+    kafkaTemplate.send("sentry-topic", message);
+    return "Message sent: " + message;
+  }
+}

sentry-samples/sentry-samples-spring-boot-4/src/main/resources/application-kafka.properties

@@ -0,0 +1,10 @@
+# Kafka — activate with: --spring.profiles.active=kafka
+sentry.enable-queue-tracing=true
+
+spring.kafka.bootstrap-servers=localhost:9092
+spring.kafka.consumer.group-id=sentry-sample-group
+spring.kafka.consumer.auto-offset-reset=earliest
+spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
+spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
+spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
+spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer