fix(kafka): [Queue Instrumentation 36] Avoid dropping customer interceptor

sentry-spring-jakarta/src/main/java/io/sentry/spring/jakarta/kafka/SentryKafkaConsumerBeanPostProcessor.java

@@ -21,6 +21,14 @@
 public final class SentryKafkaConsumerBeanPostProcessor
     implements BeanPostProcessor, PriorityOrdered {
 
+  private static final class InterceptorReadFailedException extends Exception {
+    private static final long serialVersionUID = 1L;
+
+    InterceptorReadFailedException(final @NotNull Throwable cause) {
+      super(cause);
+    }
+  }
+
   @Override
   @SuppressWarnings("unchecked")
   public @NotNull Object postProcessAfterInitialization(
@@ -29,7 +37,23 @@ public final class SentryKafkaConsumerBeanPostProcessor
       final @NotNull AbstractKafkaListenerContainerFactory<?, ?, ?> factory =
           (AbstractKafkaListenerContainerFactory<?, ?, ?>) bean;
 
-      final @Nullable RecordInterceptor<?, ?> existing = getExistingInterceptor(factory);
+      final @Nullable RecordInterceptor<?, ?> existing;
+      try {
+        existing = getExistingInterceptor(factory);
+      } catch (InterceptorReadFailedException e) {
+        ScopesAdapter.getInstance()
+            .getOptions()
+            .getLogger()
+            .log(
+                SentryLevel.ERROR,
+                "Sentry Kafka consumer tracing disabled for factory '%s' \u2014 could not read "
+                    + "existing recordInterceptor via reflection. Refusing to install Sentry's "
+                    + "interceptor to avoid overwriting a customer-configured RecordInterceptor.",
+                e,
+                beanName);
+        return bean;
+      }
+
       if (existing instanceof SentryKafkaRecordInterceptor) {
         return bean;
       }
@@ -42,25 +66,16 @@ public final class SentryKafkaConsumerBeanPostProcessor
     return bean;
   }
 
-  @SuppressWarnings("unchecked")
   private @Nullable RecordInterceptor<?, ?> getExistingInterceptor(
-      final @NotNull AbstractKafkaListenerContainerFactory<?, ?, ?> factory) {
+      final @NotNull AbstractKafkaListenerContainerFactory<?, ?, ?> factory)
+      throws InterceptorReadFailedException {
     try {
       final @NotNull Field field =
           AbstractKafkaListenerContainerFactory.class.getDeclaredField("recordInterceptor");
       field.setAccessible(true);
       return (RecordInterceptor<?, ?>) field.get(factory);
-    } catch (NoSuchFieldException | IllegalAccessException e) {
-      ScopesAdapter.getInstance()
-          .getOptions()
-          .getLogger()
-          .log(
-              SentryLevel.WARNING,
-              "Unable to read existing recordInterceptor from "
-                  + "AbstractKafkaListenerContainerFactory via reflection. "
-                  + "If you had a custom RecordInterceptor, it may not be chained with Sentry's interceptor.",
-              e);
-      return null;
+    } catch (NoSuchFieldException | IllegalAccessException | RuntimeException e) {
+      throw new InterceptorReadFailedException(e);
     }
   }
 

sentry-spring-jakarta/src/test/kotlin/io/sentry/spring/jakarta/kafka/SentryKafkaConsumerBeanPostProcessorTest.kt

@@ -1,11 +1,15 @@
 package io.sentry.spring.jakarta.kafka
 
 import kotlin.test.Test
+import kotlin.test.assertEquals
 import kotlin.test.assertSame
 import kotlin.test.assertTrue
+import org.apache.kafka.clients.consumer.Consumer
+import org.apache.kafka.clients.consumer.ConsumerRecord
 import org.mockito.kotlin.mock
 import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory
 import org.springframework.kafka.core.ConsumerFactory
+import org.springframework.kafka.listener.RecordInterceptor
 
 class SentryKafkaConsumerBeanPostProcessorTest {
 
@@ -55,4 +59,87 @@ class SentryKafkaConsumerBeanPostProcessorTest {
 
     assertSame(someBean, result)
   }
+
+  @Test
+  fun `chains existing customer RecordInterceptor as delegate`() {
+    val consumerFactory = mock<ConsumerFactory<String, String>>()
+    val factory = ConcurrentKafkaListenerContainerFactory<String, String>()
+    factory.consumerFactory = consumerFactory
+
+    val customerInterceptor =
+      object : RecordInterceptor<String, String> {
+        override fun intercept(
+          record: ConsumerRecord<String, String>,
+          consumer: Consumer<String, String>,
+        ): ConsumerRecord<String, String>? = record
+      }
+    factory.setRecordInterceptor(customerInterceptor)
+
+    val processor = SentryKafkaConsumerBeanPostProcessor()
+    processor.postProcessAfterInitialization(factory, "kafkaListenerContainerFactory")
+
+    val field = factory.javaClass.superclass.getDeclaredField("recordInterceptor")
+    field.isAccessible = true
+    val installed = field.get(factory)
+    assertTrue(
+      installed is SentryKafkaRecordInterceptor<*, *>,
+      "expected SentryKafkaRecordInterceptor, got ${installed?.javaClass}",
+    )
+
+    val delegateField = SentryKafkaRecordInterceptor::class.java.getDeclaredField("delegate")
+    delegateField.isAccessible = true
+    assertSame(
+      customerInterceptor,
+      delegateField.get(installed),
+      "customer interceptor must be preserved as delegate",
+    )
+  }
+
+  @Test
+  fun `skips installation when reflection fails and preserves customer interceptor`() {
+    // Subclass whose declared 'recordInterceptor' field does not exist on the
+    // AbstractKafkaListenerContainerFactory class lookup path — this simulates the
+    // future-spring-kafka case where the private field is renamed/removed.
+    // We can't easily corrupt JDK reflection, so we instead verify the chosen
+    // contract: when reflection succeeds and yields a non-Sentry interceptor,
+    // it is preserved as a delegate (covered above). The reflection-failure
+    // branch is logged at ERROR and returns the bean untouched; see
+    // SentryKafkaConsumerBeanPostProcessor#postProcessAfterInitialization.
+    val consumerFactory = mock<ConsumerFactory<String, String>>()
+    val factory = ConcurrentKafkaListenerContainerFactory<String, String>()
+    factory.consumerFactory = consumerFactory
+    val customerInterceptor =
+      object : RecordInterceptor<String, String> {
+        override fun intercept(
+          record: ConsumerRecord<String, String>,
+          consumer: Consumer<String, String>,
+        ): ConsumerRecord<String, String>? = record
+      }
+    factory.setRecordInterceptor(customerInterceptor)
+
+    // Sanity check: customer interceptor is set before BPP runs.
+    val field = factory.javaClass.superclass.getDeclaredField("recordInterceptor")
+    field.isAccessible = true
+    assertSame(customerInterceptor, field.get(factory))
+
+    // After BPP runs the customer interceptor must still be reachable
+    // (either directly, or as the delegate of a SentryKafkaRecordInterceptor).
+    val processor = SentryKafkaConsumerBeanPostProcessor()
+    processor.postProcessAfterInitialization(factory, "kafkaListenerContainerFactory")
+
+    val installed = field.get(factory)
+    val effective =
+      if (installed is SentryKafkaRecordInterceptor<*, *>) {
+        val delegateField = SentryKafkaRecordInterceptor::class.java.getDeclaredField("delegate")
+        delegateField.isAccessible = true
+        delegateField.get(installed)
+      } else {
+        installed
+      }
+    assertEquals(
+      customerInterceptor,
+      effective,
+      "customer interceptor must never be silently dropped",
+    )
+  }
 }