vertx-kafka-client-3.6

Author: SylvainJuge

instrumentation/vertx/vertx-kafka-client-3.6/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/vertx/kafka/v3_6/KafkaReadStreamImplInstrumentation.java

@@ -16,6 +16,8 @@
 import io.vertx.core.Handler;
 import io.vertx.kafka.client.consumer.impl.KafkaReadStreamImpl;
 import net.bytebuddy.asm.Advice;
+import net.bytebuddy.asm.Advice.AssignReturned;
+import net.bytebuddy.asm.Advice.AssignReturned.ToArguments.ToArgument;
 import net.bytebuddy.description.type.TypeDescription;
 import net.bytebuddy.matcher.ElementMatcher;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -47,24 +49,26 @@ public void transform(TypeTransformer transformer) {
   @SuppressWarnings("unused")
   public static class HandlerAdvice {
 
+    @AssignReturned.ToArguments(@ToArgument(0))
     @Advice.OnMethodEnter(suppress = Throwable.class)
-    public static <K, V> void onEnter(
+    public static <K, V> Handler<ConsumerRecord<K, V>> onEnter(
         @Advice.This KafkaReadStreamImpl<K, V> readStream,
-        @Advice.Argument(value = 0, readOnly = false) Handler<ConsumerRecord<K, V>> handler) {
+        @Advice.Argument(0) Handler<ConsumerRecord<K, V>> handler) {
 
-      handler = new InstrumentedSingleRecordHandler<>(handler);
+      return new InstrumentedSingleRecordHandler<>(handler);
     }
   }
 
   @SuppressWarnings("unused")
   public static class BatchHandlerAdvice {
 
+    @AssignReturned.ToArguments(@ToArgument(0))
     @Advice.OnMethodEnter(suppress = Throwable.class)
-    public static <K, V> void onEnter(
+    public static <K, V> Handler<ConsumerRecords<K, V>> onEnter(
         @Advice.This KafkaReadStreamImpl<K, V> readStream,
-        @Advice.Argument(value = 0, readOnly = false) Handler<ConsumerRecords<K, V>> handler) {
+        @Advice.Argument(0) Handler<ConsumerRecords<K, V>> handler) {
 
-      handler = new InstrumentedBatchRecordsHandler<>(handler);
+      return new InstrumentedBatchRecordsHandler<>(handler);
     }
   }
 

instrumentation/vertx/vertx-kafka-client-3.6/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/vertx/kafka/v3_6/VertxKafkaInstrumentationModule.java

@@ -10,10 +10,12 @@
 import com.google.auto.service.AutoService;
 import io.opentelemetry.javaagent.extension.instrumentation.InstrumentationModule;
 import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation;
+import io.opentelemetry.javaagent.extension.instrumentation.internal.ExperimentalInstrumentationModule;
 import java.util.List;
 
 @AutoService(InstrumentationModule.class)
-public class VertxKafkaInstrumentationModule extends InstrumentationModule {
+public class VertxKafkaInstrumentationModule extends InstrumentationModule
+    implements ExperimentalInstrumentationModule {
 
   public VertxKafkaInstrumentationModule() {
     super("vertx-kafka-client", "vertx-kafka-client-3.6", "vertx");
@@ -24,4 +26,9 @@ public List<TypeInstrumentation> typeInstrumentations() {
     return asList(
         new KafkaReadStreamImplInstrumentation(), new KafkaConsumerRecordsImplInstrumentation());
   }
+
+  @Override
+  public boolean isIndyReady() {
+    return true;
+  }
 }