Fix spring kafka interceptor wrappers not delegating some methods (#10935)

instrumentation/spring/spring-kafka-2.7/library/build.gradle.kts

@@ -10,7 +10,8 @@ dependencies {
 
   implementation(project(":instrumentation:kafka:kafka-clients:kafka-clients-common:library"))
 
-  compileOnly("org.springframework.kafka:spring-kafka:2.7.0")
+  // compiling against 2.8.0 to use methods that are not present in 2.7
+  compileOnly("org.springframework.kafka:spring-kafka:2.8.0")
 
   testImplementation(project(":instrumentation:spring:spring-kafka-2.7:testing"))
   testImplementation(project(":instrumentation:kafka:kafka-clients:kafka-clients-2.6:library"))

instrumentation/spring/spring-kafka-2.7/library/src/main/java/io/opentelemetry/instrumentation/spring/kafka/v2_7/InstrumentedBatchInterceptor.java

@@ -12,6 +12,7 @@ import io.opentelemetry.instrumentation.api.util.VirtualField;
 import io.opentelemetry.instrumentation.kafka.internal.KafkaConsumerContext;
 import io.opentelemetry.instrumentation.kafka.internal.KafkaConsumerContextUtil;
 import io.opentelemetry.instrumentation.kafka.internal.KafkaReceiveRequest;
+import io.opentelemetry.javaagent.tooling.muzzle.NoMuzzle;
 import java.lang.ref.WeakReference;
 import javax.annotation.Nullable;
 import org.apache.kafka.clients.consumer.Consumer;
@@ -94,4 +95,20 @@ final class InstrumentedBatchInterceptor<K, V> implements BatchInterceptor<K, V>
       lastProcessed.set(new WeakReference<>(records));
     }
   }
+
+  @NoMuzzle // method was added in 2.8.0
+  @Override
+  public void setupThreadState(Consumer<?, ?> consumer) {
+    if (decorated != null) {
+      decorated.setupThreadState(consumer);
+    }
+  }
+
+  @NoMuzzle // method was added in 2.8.0
+  @Override
+  public void clearThreadState(Consumer<?, ?> consumer) {
+    if (decorated != null) {
+      decorated.clearThreadState(consumer);
+    }
+  }
 }

instrumentation/spring/spring-kafka-2.7/library/src/main/java/io/opentelemetry/instrumentation/spring/kafka/v2_7/InstrumentedRecordInterceptor.java

@@ -92,4 +92,28 @@ final class InstrumentedRecordInterceptor<K, V> implements RecordInterceptor<K,
       processInstrumenter.end(state.context(), request, null, error);
     }
   }
+
+  @NoMuzzle // method was added in 2.8.0
+  @Override
+  public void afterRecord(ConsumerRecord<K, V> record, Consumer<K, V> consumer) {
+    if (decorated != null) {
+      decorated.afterRecord(record, consumer);
+    }
+  }
+
+  @NoMuzzle // method was added in 2.8.0
+  @Override
+  public void setupThreadState(Consumer<?, ?> consumer) {
+    if (decorated != null) {
+      decorated.setupThreadState(consumer);
+    }
+  }
+
+  @NoMuzzle // method was added in 2.8.0
+  @Override
+  public void clearThreadState(Consumer<?, ?> consumer) {
+    if (decorated != null) {
+      decorated.clearThreadState(consumer);
+    }
+  }
 }