From bfc5ec35ceeffa0b350be649990a3042f181b5fc Mon Sep 17 00:00:00 2001
From: OpenTelemetry Bot <107717825+opentelemetrybot@users.noreply.github.com>
Date: Fri, 14 Jun 2024 10:29:24 +0200
Subject: [PATCH] [release/v1.33.x] Fix InstrumentedRecordInterceptor closing
 the trace too early (#11592)

Co-authored-by: Ben <luaworfox@gmail.com>
---
 .../v2_7/InstrumentedBatchInterceptor.java     | 18 ++++++++++++------
 .../v2_7/InstrumentedRecordInterceptor.java    | 18 ++++++++++++------
 2 files changed, 24 insertions(+), 12 deletions(-)

diff --git a/instrumentation/spring/spring-kafka-2.7/library/src/main/java/io/opentelemetry/instrumentation/spring/kafka/v2_7/InstrumentedBatchInterceptor.java b/instrumentation/spring/spring-kafka-2.7/library/src/main/java/io/opentelemetry/instrumentation/spring/kafka/v2_7/InstrumentedBatchInterceptor.java
index 4d30d8c97b15..a7bd18cd04d2 100644
--- a/instrumentation/spring/spring-kafka-2.7/library/src/main/java/io/opentelemetry/instrumentation/spring/kafka/v2_7/InstrumentedBatchInterceptor.java
+++ b/instrumentation/spring/spring-kafka-2.7/library/src/main/java/io/opentelemetry/instrumentation/spring/kafka/v2_7/InstrumentedBatchInterceptor.java
@@ -56,17 +56,23 @@ private static Context getParentContext(ConsumerRecords<?, ?> records) {
 
   @Override
   public void success(ConsumerRecords<K, V> records, Consumer<K, V> consumer) {
-    end(records, null);
-    if (decorated != null) {
-      decorated.success(records, consumer);
+    try {
+      if (decorated != null) {
+        decorated.success(records, consumer);
+      }
+    } finally {
+      end(records, null);
     }
   }
 
   @Override
   public void failure(ConsumerRecords<K, V> records, Exception exception, Consumer<K, V> consumer) {
-    end(records, exception);
-    if (decorated != null) {
-      decorated.failure(records, exception, consumer);
+    try {
+      if (decorated != null) {
+        decorated.failure(records, exception, consumer);
+      }
+    } finally {
+      end(records, exception);
     }
   }
 
diff --git a/instrumentation/spring/spring-kafka-2.7/library/src/main/java/io/opentelemetry/instrumentation/spring/kafka/v2_7/InstrumentedRecordInterceptor.java b/instrumentation/spring/spring-kafka-2.7/library/src/main/java/io/opentelemetry/instrumentation/spring/kafka/v2_7/InstrumentedRecordInterceptor.java
index 992a91d93701..9cf9a945dae0 100644
--- a/instrumentation/spring/spring-kafka-2.7/library/src/main/java/io/opentelemetry/instrumentation/spring/kafka/v2_7/InstrumentedRecordInterceptor.java
+++ b/instrumentation/spring/spring-kafka-2.7/library/src/main/java/io/opentelemetry/instrumentation/spring/kafka/v2_7/InstrumentedRecordInterceptor.java
@@ -69,17 +69,23 @@ private static Context getParentContext(ConsumerRecord<?, ?> record) {
 
   @Override
   public void success(ConsumerRecord<K, V> record, Consumer<K, V> consumer) {
-    end(record, null);
-    if (decorated != null) {
-      decorated.success(record, consumer);
+    try {
+      if (decorated != null) {
+        decorated.success(record, consumer);
+      }
+    } finally {
+      end(record, null);
     }
   }
 
   @Override
   public void failure(ConsumerRecord<K, V> record, Exception exception, Consumer<K, V> consumer) {
-    end(record, exception);
-    if (decorated != null) {
-      decorated.failure(record, exception, consumer);
+    try {
+      if (decorated != null) {
+        decorated.failure(record, exception, consumer);
+      }
+    } finally {
+      end(record, exception);
     }
   }