feat(kafka): Add consumer demo to console sample

Show end-to-end Kafka queue tracing in the console sample by running a background consumer thread, producing a message, and waiting for consume before exit.\n\nAdd a no-arg constructor to SentryKafkaConsumerInterceptor so kafka-clients can instantiate it from interceptor.classes, and add test coverage for that constructor.

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

Author: adinauer

sentry-kafka/api/sentry-kafka.api

@@ -5,6 +5,7 @@ public final class io/sentry/kafka/BuildConfig {
 
 public final class io/sentry/kafka/SentryKafkaConsumerInterceptor : org/apache/kafka/clients/consumer/ConsumerInterceptor {
 	public static final field TRACE_ORIGIN Ljava/lang/String;
+	public fun <init> ()V
 	public fun <init> (Lio/sentry/IScopes;)V
 	public fun close ()V
 	public fun configure (Ljava/util/Map;)V

sentry-kafka/src/main/java/io/sentry/kafka/SentryKafkaConsumerInterceptor.java

@@ -3,6 +3,7 @@
 import io.sentry.BaggageHeader;
 import io.sentry.IScopes;
 import io.sentry.ITransaction;
+import io.sentry.ScopesAdapter;
 import io.sentry.SentryTraceHeader;
 import io.sentry.SpanDataConvention;
 import io.sentry.SpanStatus;
@@ -29,6 +30,10 @@ public final class SentryKafkaConsumerInterceptor<K, V> implements ConsumerInter
 
   private final @NotNull IScopes scopes;
 
+  public SentryKafkaConsumerInterceptor() {
+    this(ScopesAdapter.getInstance());
+  }
+
   public SentryKafkaConsumerInterceptor(final @NotNull IScopes scopes) {
     this.scopes = scopes;
   }

sentry-kafka/src/test/kotlin/io/sentry/kafka/SentryKafkaConsumerInterceptorTest.kt

@@ -2,9 +2,13 @@ package io.sentry.kafka
 
 import io.sentry.IScopes
 import io.sentry.ITransaction
+import io.sentry.Sentry
 import io.sentry.SentryOptions
 import io.sentry.TransactionContext
 import io.sentry.TransactionOptions
+import io.sentry.test.initForTest
+import kotlin.test.AfterTest
+import kotlin.test.BeforeTest
 import kotlin.test.Test
 import kotlin.test.assertSame
 import org.apache.kafka.clients.consumer.ConsumerRecord
@@ -19,6 +23,20 @@ import org.mockito.kotlin.whenever
 
 class SentryKafkaConsumerInterceptorTest {
 
+  @BeforeTest
+  fun setup() {
+    initForTest {
+      it.dsn = "https://key@sentry.io/proj"
+      it.isEnableQueueTracing = true
+      it.tracesSampleRate = 1.0
+    }
+  }
+
+  @AfterTest
+  fun teardown() {
+    Sentry.close()
+  }
+
   @Test
   fun `does nothing when queue tracing is disabled`() {
     val scopes = mock<IScopes>()
@@ -64,6 +82,16 @@ class SentryKafkaConsumerInterceptorTest {
     interceptor.onCommit(mapOf(TopicPartition("my-topic", 0) to OffsetAndMetadata(1)))
   }
 
+  @Test
+  fun `no-arg constructor uses current scopes`() {
+    val interceptor = SentryKafkaConsumerInterceptor<String, String>()
+    val records = singleRecordBatch()
+
+    val result = interceptor.onConsume(records)
+
+    assertSame(records, result)
+  }
+
   private fun singleRecordBatch(): ConsumerRecords<String, String> {
     val partition = TopicPartition("my-topic", 0)
     val record = ConsumerRecord("my-topic", 0, 0L, "key", "value")

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

@@ -3,18 +3,27 @@
 import io.sentry.*;
 import io.sentry.clientreport.DiscardReason;
 import io.sentry.jcache.SentryJCacheWrapper;
+import io.sentry.kafka.SentryKafkaConsumerInterceptor;
 import io.sentry.kafka.SentryKafkaProducerInterceptor;
 import io.sentry.protocol.Message;
 import io.sentry.protocol.User;
+import java.time.Duration;
 import java.util.Collections;
 import java.util.Properties;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 import javax.cache.Cache;
 import javax.cache.CacheManager;
 import javax.cache.Caching;
 import javax.cache.configuration.MutableConfiguration;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.serialization.StringDeserializer;
 import org.apache.kafka.common.serialization.StringSerializer;
 
 public class Main {
@@ -261,6 +270,10 @@ private static void captureMetrics() {
   }
 
   private static void demonstrateKafkaTracing() {
+    final String topic = "sentry-topic-console-sample";
+    final CountDownLatch consumedLatch = new CountDownLatch(1);
+    final Thread consumerThread = startKafkaConsumerThread(topic, consumedLatch);
+
     final Properties producerProperties = new Properties();
     producerProperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
     producerProperties.put(
@@ -273,17 +286,70 @@ private static void demonstrateKafkaTracing() {
     final ITransaction transaction = Sentry.startTransaction("kafka-demo", "demo");
     try (ISentryLifecycleToken ignored = transaction.makeCurrent()) {
       try (KafkaProducer<String, String> producer = new KafkaProducer<>(producerProperties)) {
-        producer.send(new ProducerRecord<>("sentry-topic", "sentry-kafka sample message")).get();
+        Thread.sleep(500);
+        producer.send(new ProducerRecord<>(topic, "sentry-kafka sample message")).get();
       } catch (InterruptedException e) {
         Thread.currentThread().interrupt();
       } catch (Exception ignoredException) {
         // local broker may not be available when running the sample
       }
+
+      try {
+        consumedLatch.await(5, TimeUnit.SECONDS);
+      } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
+      }
     } finally {
+      consumerThread.interrupt();
+      try {
+        consumerThread.join(1000);
+      } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
+      }
       transaction.finish();
     }
   }
 
+  private static Thread startKafkaConsumerThread(
+      final String topic, final CountDownLatch consumedLatch) {
+    final Thread consumerThread =
+        new Thread(
+            () -> {
+              final Properties consumerProperties = new Properties();
+              consumerProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
+              consumerProperties.put(
+                  ConsumerConfig.GROUP_ID_CONFIG, "sentry-console-sample-" + UUID.randomUUID());
+              consumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+              consumerProperties.put(
+                  ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
+              consumerProperties.put(
+                  ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
+                  StringDeserializer.class.getName());
+              consumerProperties.put(
+                  ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG,
+                  SentryKafkaConsumerInterceptor.class.getName());
+
+              try (KafkaConsumer<String, String> consumer =
+                  new KafkaConsumer<>(consumerProperties)) {
+                consumer.subscribe(Collections.singletonList(topic));
+
+                while (!Thread.currentThread().isInterrupted() && consumedLatch.getCount() > 0) {
+                  final ConsumerRecords<String, String> records =
+                      consumer.poll(Duration.ofMillis(500));
+                  if (!records.isEmpty()) {
+                    consumedLatch.countDown();
+                    break;
+                  }
+                }
+              } catch (Exception ignored) {
+                // local broker may not be available when running the sample
+              }
+            },
+            "sentry-kafka-sample-consumer");
+    consumerThread.start();
+    return consumerThread;
+  }
+
   private static class SomeEventProcessor implements EventProcessor {
     @Override
     public SentryEvent process(SentryEvent event, Hint hint) {