feat(samples): Add opt-in Kafka console e2e coverage

Gate the console Kafka showcase behind SENTRY_SAMPLE_KAFKA_BOOTSTRAP_SERVERS so Kafka behavior is enabled only when configured. Keep the showcase isolated in KafkaShowcase and use fail-fast Kafka client timeouts for local runs.\n\nExtend console system tests to assert producer and consumer queue tracing when Kafka is enabled. Update system-test-runner to provision or reuse a local Kafka broker for the console module and clean up runner-managed resources.

Co-Authored-By: Claude <noreply@anthropic.com>

Author: adinauer

sentry-samples/sentry-samples-console/src/main/java/io/sentry/samples/console/KafkaShowcase.java

@@ -24,19 +24,22 @@ final class KafkaShowcase {
 
   private KafkaShowcase() {}
 
-  static void demonstrate() {
+  static void demonstrate(final String bootstrapServers) {
     final String topic = "sentry-topic-console-sample";
     final CountDownLatch consumedLatch = new CountDownLatch(1);
-    final Thread consumerThread = startKafkaConsumerThread(topic, consumedLatch);
+    final Thread consumerThread = startKafkaConsumerThread(topic, bootstrapServers, consumedLatch);
 
     final Properties producerProperties = new Properties();
-    producerProperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
+    producerProperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
     producerProperties.put(
         ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
     producerProperties.put(
         ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
     producerProperties.put(
         ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, SentryKafkaProducerInterceptor.class.getName());
+    producerProperties.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 2000);
+    producerProperties.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 2000);
+    producerProperties.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 3000);
 
     final ITransaction transaction = Sentry.startTransaction("kafka-demo", "demo");
     try (ISentryLifecycleToken ignored = transaction.makeCurrent()) {
@@ -66,12 +69,12 @@ static void demonstrate() {
   }
 
   private static Thread startKafkaConsumerThread(
-      final String topic, final CountDownLatch consumedLatch) {
+      final String topic, final String bootstrapServers, final CountDownLatch consumedLatch) {
     final Thread consumerThread =
         new Thread(
             () -> {
               final Properties consumerProperties = new Properties();
-              consumerProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
+              consumerProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
               consumerProperties.put(
                   ConsumerConfig.GROUP_ID_CONFIG, "sentry-console-sample-" + UUID.randomUUID());
               consumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
@@ -83,6 +86,8 @@ private static Thread startKafkaConsumerThread(
               consumerProperties.put(
                   ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG,
                   SentryKafkaConsumerInterceptor.class.getName());
+              consumerProperties.put(ConsumerConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, 2000);
+              consumerProperties.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, 2000);
 
               try (KafkaConsumer<String, String> consumer =
                   new KafkaConsumer<>(consumerProperties)) {

sentry-samples/sentry-samples-console/src/main/java/io/sentry/samples/console/Main.java

@@ -16,6 +16,10 @@ public class Main {
   private static long numberOfDiscardedSpansDueToOverflow = 0;
 
   public static void main(String[] args) throws InterruptedException {
+    final String kafkaBootstrapServers = System.getenv("SENTRY_SAMPLE_KAFKA_BOOTSTRAP_SERVERS");
+    final boolean kafkaEnabled =
+        kafkaBootstrapServers != null && !kafkaBootstrapServers.trim().isEmpty();
+
     Sentry.init(
         options -> {
           // NOTE: Replace the test DSN below with YOUR OWN DSN to see the events from this app in
@@ -95,7 +99,7 @@ public static void main(String[] args) throws InterruptedException {
 
           // Enable cache tracing to create spans for cache operations
           options.setEnableCacheTracing(true);
-          options.setEnableQueueTracing(true);
+          options.setEnableQueueTracing(kafkaEnabled);
 
           // Determine traces sample rate based on the sampling context
           //          options.setTracesSampler(
@@ -181,9 +185,10 @@ public static void main(String[] args) throws InterruptedException {
 
     // Kafka queue tracing with kafka-clients interceptors.
     //
-    // This uses the native producer interceptor from sentry-kafka.
-    // If no local Kafka broker is available, this block exits quietly.
-    KafkaShowcase.demonstrate();
+    // Enable with: SENTRY_SAMPLE_KAFKA_BOOTSTRAP_SERVERS=localhost:9092
+    if (kafkaEnabled) {
+      KafkaShowcase.demonstrate(kafkaBootstrapServers);
+    }
 
     // Performance feature
     //

sentry-samples/sentry-samples-console/src/test/kotlin/io/sentry/systemtest/ConsoleApplicationSystemTest.kt

@@ -19,19 +19,7 @@ class ConsoleApplicationSystemTest {
 
   @Test
   fun `console application sends expected events when run as JAR`() {
-    val jarFile = testHelper.findJar("sentry-samples-console")
-    val process =
-      testHelper.launch(
-        jarFile,
-        mapOf(
-          "SENTRY_DSN" to testHelper.dsn,
-          "SENTRY_TRACES_SAMPLE_RATE" to "1.0",
-          "SENTRY_ENABLE_PRETTY_SERIALIZATION_OUTPUT" to "false",
-          "SENTRY_DEBUG" to "true",
-          "SENTRY_PROFILE_SESSION_SAMPLE_RATE" to "1.0",
-          "SENTRY_PROFILE_LIFECYCLE" to "TRACE",
-        ),
-      )
+    val process = launchConsoleProcess()
 
     process.waitFor(30, TimeUnit.SECONDS)
     assertEquals(0, process.exitValue())
@@ -40,6 +28,40 @@ class ConsoleApplicationSystemTest {
     verifyExpectedEvents()
   }
 
+  @Test
+  fun `console application sends kafka producer and consumer tracing when kafka is enabled`() {
+    val process =
+      launchConsoleProcess(mapOf("SENTRY_SAMPLE_KAFKA_BOOTSTRAP_SERVERS" to "localhost:9092"))
+
+    process.waitFor(30, TimeUnit.SECONDS)
+    assertEquals(0, process.exitValue())
+
+    testHelper.ensureTransactionReceived { transaction, _ ->
+      transaction.transaction == "kafka-demo" &&
+        testHelper.doesTransactionContainSpanWithOp(transaction, "queue.publish")
+    }
+
+    testHelper.ensureTransactionReceived { transaction, _ ->
+      testHelper.doesTransactionHaveOp(transaction, "queue.receive") &&
+        transaction.contexts.trace?.data?.get("messaging.system") == "kafka"
+    }
+  }
+
+  private fun launchConsoleProcess(overrides: Map<String, String> = emptyMap()): Process {
+    val jarFile = testHelper.findJar("sentry-samples-console")
+    val env =
+      mutableMapOf(
+        "SENTRY_DSN" to testHelper.dsn,
+        "SENTRY_TRACES_SAMPLE_RATE" to "1.0",
+        "SENTRY_ENABLE_PRETTY_SERIALIZATION_OUTPUT" to "false",
+        "SENTRY_DEBUG" to "true",
+        "SENTRY_PROFILE_SESSION_SAMPLE_RATE" to "1.0",
+        "SENTRY_PROFILE_LIFECYCLE" to "TRACE",
+      )
+    env.putAll(overrides)
+    return testHelper.launch(jarFile, env)
+  }
+
   private fun verifyExpectedEvents() {
     var profilerId: SentryId? = null
     // Verify we received a "Fatal message!" event

test/system-test-runner.py

@@ -42,6 +42,7 @@
 import argparse
 import requests
 import threading
+import socket
 from pathlib import Path
 from typing import Optional, List, Tuple
 from dataclasses import dataclass
@@ -65,6 +66,9 @@
     "SENTRY_ENABLE_CACHE_TRACING": "true"
 }
 
+KAFKA_CONTAINER_NAME = "sentry-java-system-test-kafka"
+KAFKA_BOOTSTRAP_SERVERS = "localhost:9092"
+
 class ServerType(Enum):
     TOMCAT = 0
     SPRING = 1
@@ -155,6 +159,7 @@ def __init__(self):
         self.mock_server = Server(name="Mock", pid_filepath="sentry-mock-server.pid")
         self.tomcat_server = Server(name="Tomcat", pid_filepath="tomcat-server.pid")
         self.spring_server = Server(name="Spring", pid_filepath="spring-server.pid")
+        self.kafka_started_by_runner = False
 
         # Load existing PIDs if available
         for server in (self.mock_server, self.tomcat_server, self.spring_server):
@@ -196,7 +201,78 @@ def kill_process(self, pid: int, name: str) -> None:
         except (OSError, ProcessLookupError):
             print(f"Process {pid} was already dead")
 
+    def module_requires_kafka(self, sample_module: str) -> bool:
+        return sample_module == "sentry-samples-console"
+
+    def wait_for_port(self, host: str, port: int, max_attempts: int = 20) -> bool:
+        for _ in range(max_attempts):
+            try:
+                with socket.create_connection((host, port), timeout=1):
+                    return True
+            except OSError:
+                time.sleep(1)
+        return False
 
+    def start_kafka_broker(self) -> None:
+        if self.wait_for_port("localhost", 9092, max_attempts=1):
+            print("Kafka broker already running on localhost:9092, reusing it.")
+            self.kafka_started_by_runner = False
+            return
+
+        self.stop_kafka_broker()
+
+        print("Starting Kafka broker (Redpanda) for system tests...")
+        run_result = subprocess.run(
+            [
+                "docker",
+                "run",
+                "-d",
+                "--name",
+                KAFKA_CONTAINER_NAME,
+                "-p",
+                "9092:9092",
+                "docker.redpanda.com/redpandadata/redpanda:v24.1.9",
+                "redpanda",
+                "start",
+                "--overprovisioned",
+                "--smp",
+                "1",
+                "--memory",
+                "1G",
+                "--reserve-memory",
+                "0M",
+                "--node-id",
+                "0",
+                "--check=false",
+                "--kafka-addr",
+                "PLAINTEXT://0.0.0.0:9092",
+                "--advertise-kafka-addr",
+                "PLAINTEXT://localhost:9092",
+            ],
+            check=False,
+            capture_output=True,
+            text=True,
+        )
+
+        if run_result.returncode != 0:
+            raise RuntimeError(f"Failed to start Kafka container: {run_result.stderr}")
+
+        if not self.wait_for_port("localhost", 9092, max_attempts=30):
+            raise RuntimeError("Kafka broker did not become ready on localhost:9092")
+
+        self.kafka_started_by_runner = True
+
+    def stop_kafka_broker(self) -> None:
+        if not self.kafka_started_by_runner:
+            return
+
+        subprocess.run(
+            ["docker", "rm", "-f", KAFKA_CONTAINER_NAME],
+            check=False,
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+        )
+        self.kafka_started_by_runner = False
 
     def start_sentry_mock_server(self) -> None:
         """Start the Sentry mock server."""
@@ -557,6 +633,12 @@ def setup_test_infrastructure(self, sample_module: str, java_agent: str,
                                  java_agent_auto_init: str, build_before_run: str,
                                  server_type: Optional[ServerType]) -> int:
         """Set up test infrastructure. Returns 0 on success, error code on failure."""
+        if self.module_requires_kafka(sample_module):
+            self.start_kafka_broker()
+            os.environ["SENTRY_SAMPLE_KAFKA_BOOTSTRAP_SERVERS"] = KAFKA_BOOTSTRAP_SERVERS
+        else:
+            os.environ.pop("SENTRY_SAMPLE_KAFKA_BOOTSTRAP_SERVERS", None)
+
         # Build if requested
         if build_before_run == "1":
             print("Building before test run")
@@ -624,6 +706,8 @@ def run_single_test(self, sample_module: str, java_agent: str,
             elif server_type == ServerType.SPRING:
                 self.stop_spring_server()
             self.stop_sentry_mock_server()
+            self.stop_kafka_broker()
+            os.environ.pop("SENTRY_SAMPLE_KAFKA_BOOTSTRAP_SERVERS", None)
 
     def run_all_tests(self) -> int:
         """Run all system tests."""
@@ -954,6 +1038,8 @@ def cleanup_on_exit(self, signum, frame):
         self.stop_spring_server()
         self.stop_sentry_mock_server()
         self.stop_tomcat_server()
+        self.stop_kafka_broker()
+        os.environ.pop("SENTRY_SAMPLE_KAFKA_BOOTSTRAP_SERVERS", None)
         sys.exit(1)
 
 def main():
@@ -1152,6 +1238,8 @@ def main():
             runner.stop_spring_server()
             runner.stop_sentry_mock_server()
             runner.stop_tomcat_server()
+            runner.stop_kafka_broker()
+            os.environ.pop("SENTRY_SAMPLE_KAFKA_BOOTSTRAP_SERVERS", None)
 
 if __name__ == "__main__":
     sys.exit(main())