fix(spring-jakarta): Close leaked Kafka interceptor scope

Store the lifecycle token in the thread-local before trace continuation or transaction startup can throw. This keeps the cleanup path reachable and closes the forked scopes even when interceptor preparation fails.

Also log the preparation failure instead of letting the interceptor break customer processing.

Author: adinauer

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

@@ -5,6 +5,7 @@
 import io.sentry.IScopes;
 import io.sentry.ISentryLifecycleToken;
 import io.sentry.ITransaction;
+import io.sentry.SentryLevel;
 import io.sentry.SentryTraceHeader;
 import io.sentry.SpanDataConvention;
 import io.sentry.SpanStatus;
@@ -57,18 +58,21 @@ public SentryKafkaRecordInterceptor(
       return delegateIntercept(record, consumer);
     }
 
-    finishStaleContext();
-
-    final @NotNull IScopes forkedScopes = scopes.forkedRootScopes("SentryKafkaRecordInterceptor");
-    final @NotNull ISentryLifecycleToken lifecycleToken = forkedScopes.makeCurrent();
-    currentContext.set(new SentryRecordContext(lifecycleToken, null));
+    try {
+      finishStaleContext();
 
-    final @Nullable TransactionContext transactionContext = continueTrace(forkedScopes, record);
+      final @NotNull IScopes forkedScopes = scopes.forkedRootScopes("SentryKafkaRecordInterceptor");
+      final @NotNull ISentryLifecycleToken lifecycleToken = forkedScopes.makeCurrent();
+      currentContext.set(new SentryRecordContext(lifecycleToken, null));
 
-    final @Nullable ITransaction transaction =
-        startTransaction(forkedScopes, record, transactionContext);
-    currentContext.set(new SentryRecordContext(lifecycleToken, transaction));
+      final @Nullable TransactionContext transactionContext = continueTrace(forkedScopes, record);
 
+      final @Nullable ITransaction transaction =
+          startTransaction(forkedScopes, record, transactionContext);
+      currentContext.set(new SentryRecordContext(lifecycleToken, transaction));
+    } catch (Throwable t) {
+      scopes.getOptions().getLogger().log(SentryLevel.ERROR, "Unable to wrap Kafka consumer.", t);
+    }
     return delegateIntercept(record, consumer);
   }