ref(kafka): Remove raw consumer interceptor

Remove the raw Kafka consumer interceptor from sentry-kafka and update the console sample to use the manual consumer tracing helper instead. Keep producer tracing on the interceptor path and move consumer tracing to explicit record processing.

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

Author: adinauer

sentry-kafka/api/sentry-kafka.api

@@ -3,16 +3,6 @@ public final class io/sentry/kafka/BuildConfig {
 	public static final field VERSION_NAME Ljava/lang/String;
 }
 
-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
-	public fun onCommit (Ljava/util/Map;)V
-	public fun onConsume (Lorg/apache/kafka/clients/consumer/ConsumerRecords;)Lorg/apache/kafka/clients/consumer/ConsumerRecords;
-}
-
 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

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

@@ -184,11 +184,11 @@ public static void main(String[] args) throws InterruptedException {
     // cache.remove, and cache.flush spans as children of the active transaction.
     demonstrateCacheTracing();
 
-    // Kafka queue tracing with kafka-clients interceptors.
+    // Kafka queue tracing with the kafka-clients producer interceptor and manual consumer tracing.
     //
     // Enable with: SENTRY_SAMPLE_KAFKA_BOOTSTRAP_SERVERS=localhost:9092
     if (kafkaEnabled) {
-      KafkaShowcase.runKafkaWithSentryInterceptors(kafkaBootstrapServers);
+      KafkaShowcase.runKafkaWithSentryTracing(kafkaBootstrapServers);
     }
 
     // Performance feature

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

@@ -3,7 +3,7 @@
 import io.sentry.ISentryLifecycleToken;
 import io.sentry.ITransaction;
 import io.sentry.Sentry;
-import io.sentry.kafka.SentryKafkaConsumerInterceptor;
+import io.sentry.kafka.SentryKafkaConsumerTracing;
 import io.sentry.kafka.SentryKafkaProducerInterceptor;
 import java.time.Duration;
 import java.util.Collections;
@@ -12,6 +12,7 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.clients.producer.KafkaProducer;
@@ -26,10 +27,9 @@ public final class KafkaShowcase {
 
   private KafkaShowcase() {}
 
-  public static void runKafkaWithSentryInterceptors(final String bootstrapServers) {
+  public static void runKafkaWithSentryTracing(final String bootstrapServers) {
     final CountDownLatch consumedLatch = new CountDownLatch(1);
-    final Thread consumerThread =
-        startConsumerWithSentryInterceptor(bootstrapServers, consumedLatch);
+    final Thread consumerThread = startConsumerWithSentryTracing(bootstrapServers, consumedLatch);
     final Properties producerProperties = createProducerPropertiesWithSentry(bootstrapServers);
 
     final ITransaction transaction = Sentry.startTransaction("kafka-demo", "demo");
@@ -79,7 +79,7 @@ public static Properties createProducerPropertiesWithSentry(final String bootstr
     return producerProperties;
   }
 
-  public static Properties createConsumerPropertiesWithSentry(final String bootstrapServers) {
+  public static Properties createConsumerProperties(final String bootstrapServers) {
     final Properties consumerProperties = new Properties();
     consumerProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
     consumerProperties.put(
@@ -90,24 +90,19 @@ public static Properties createConsumerPropertiesWithSentry(final String bootstr
     consumerProperties.put(
         ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
 
-    // Required for Sentry queue tracing in kafka-clients consumer setup.
-    consumerProperties.put(
-        ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, SentryKafkaConsumerInterceptor.class.getName());
-
     // Optional tuning for sample stability in CI/local runs.
     consumerProperties.put(ConsumerConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, 2000);
     consumerProperties.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, 2000);
 
     return consumerProperties;
   }
 
-  private static Thread startConsumerWithSentryInterceptor(
+  private static Thread startConsumerWithSentryTracing(
       final String bootstrapServers, final CountDownLatch consumedLatch) {
     final Thread consumerThread =
         new Thread(
             () -> {
-              final Properties consumerProperties =
-                  createConsumerPropertiesWithSentry(bootstrapServers);
+              final Properties consumerProperties = createConsumerProperties(bootstrapServers);
 
               try (KafkaConsumer<String, String> consumer =
                   new KafkaConsumer<>(consumerProperties)) {
@@ -116,9 +111,11 @@ private static Thread startConsumerWithSentryInterceptor(
                 while (!Thread.currentThread().isInterrupted() && consumedLatch.getCount() > 0) {
                   final ConsumerRecords<String, String> records =
                       consumer.poll(Duration.ofMillis(500));
-                  if (!records.isEmpty()) {
-                    consumedLatch.countDown();
-                    break;
+                  for (final ConsumerRecord<String, String> record : records) {
+                    SentryKafkaConsumerTracing.withTracing(record, consumedLatch::countDown);
+                    if (consumedLatch.getCount() == 0) {
+                      break;
+                    }
                   }
                 }
               } catch (Exception ignored) {