Merge branch 'main' of github.com:jaydeluca/opentelemetry-java-instrumentation into metadata-functions

Author: jaydeluca

.github/workflows/overhead-benchmark-daily.yml

@@ -28,9 +28,18 @@ jobs:
         run: |
           rsync -avv gh-pages/benchmark-overhead/results/ benchmark-overhead/results/
 
+      - name: Set up JDK for running Gradle
+        uses: actions/setup-java@dded0888837ed1f317902acf8a20df0ad188d165 # v5.0.0
+        with:
+          distribution: temurin
+          java-version-file: .java-version
+
       - name: Setup Gradle
         uses: gradle/actions/setup-gradle@ed408507eac070d1f99cc633dbcf757c94c7933a # v4.4.3
 
+      - name: Build Latest Snapshot JAR
+        run: ./gradlew assemble -x javadoc
+
       - name: Run tests
         working-directory: benchmark-overhead
         run: ./gradlew test

benchmark-overhead/src/test/java/io/opentelemetry/agents/LatestAgentSnapshotResolver.java

@@ -5,115 +5,80 @@
 
 package io.opentelemetry.agents;
 
-import static org.joox.JOOX.$;
-
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
-import java.nio.file.StandardOpenOption;
+import java.nio.file.StandardCopyOption;
 import java.util.Optional;
-import java.time.Duration;
-import okhttp3.OkHttpClient;
-import okhttp3.Request;
-import okhttp3.Response;
-import okhttp3.ResponseBody;
-import org.joox.Match;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.w3c.dom.Document;
 
 public class LatestAgentSnapshotResolver {
 
   private static final Logger logger = LoggerFactory.getLogger(LatestAgentSnapshotResolver.class);
 
-  static final String BASE_URL =
-      "https://oss.sonatype.org/content/repositories/snapshots/io/opentelemetry/javaagent/opentelemetry-javaagent";
-  static final String LATEST_SNAPSHOT_META = BASE_URL + "/maven-metadata.xml";
-
-  private static final OkHttpClient client = new OkHttpClient.Builder()
-      .connectTimeout(Duration.ofMinutes(1))
-      .readTimeout(Duration.ofMinutes(1))
-      .build();
-
   Optional<Path> resolve() throws IOException {
-    String version = fetchLatestSnapshotVersion();
-    logger.info("Latest snapshot version is {}", version);
-    String latestFilename = fetchLatestFilename(version);
-    String url = BASE_URL + "/" + version + "/" + latestFilename;
-    byte[] jarBytes = fetchBodyBytesFrom(url);
-    Path path = Paths.get(".", "opentelemetry-javaagent-SNAPSHOT.jar");
-    Files.write(
-        path,
-        jarBytes,
-        StandardOpenOption.CREATE,
-        StandardOpenOption.WRITE,
-        StandardOpenOption.TRUNCATE_EXISTING);
-    return Optional.of(path);
-  }
+    Path localJavaagentPath = findLocalJavaagentJar();
 
-  private String fetchLatestFilename(String version) throws IOException {
-    String url = BASE_URL + "/" + version + "/maven-metadata.xml";
-    String body = fetchBodyStringFrom(url);
-    Document document = $(body).document();
-    Match match = $(document).xpath("/metadata/versioning/snapshotVersions/snapshotVersion");
-    return match.get().stream()
-        .filter(
-            elem -> {
-              Match classifierMatch = $(elem).child("classifier");
-              String classifier = classifierMatch == null ? null : classifierMatch.content();
-              String extension = $(elem).child("extension").content();
-              return "jar".equals(extension) && (classifier == null);
-            })
-        .map(e -> $(e).child("value").content())
-        .findFirst()
-        .map(value -> "opentelemetry-javaagent-" + value + ".jar")
-        .orElseThrow();
-  }
+    if (localJavaagentPath == null || !Files.exists(localJavaagentPath)) {
+      throw new IOException("Local javaagent JAR not found. Please run './gradlew :javaagent:assemble' from the project root first.");
+    }
 
-  private String fetchLatestSnapshotVersion() throws IOException {
-    String url = LATEST_SNAPSHOT_META;
-    String body = fetchBodyStringFrom(url);
-    Document document = $(body).document();
-    Match match = $(document).xpath("/metadata/versioning/latest");
-    return match.get(0).getTextContent();
-  }
+    logger.info("Using local javaagent JAR: {}", localJavaagentPath);
 
-  private String fetchBodyStringFrom(String url) throws IOException {
-    return fetchBodyFrom(url).string();
+    Path targetPath = Paths.get(".", "opentelemetry-javaagent-SNAPSHOT.jar");
+    Files.copy(localJavaagentPath, targetPath, StandardCopyOption.REPLACE_EXISTING);
+    return Optional.of(targetPath);
   }
 
-  private byte[] fetchBodyBytesFrom(String url) throws IOException {
-    return fetchBodyFrom(url).bytes();
-  }
+  private Path findLocalJavaagentJar() {
+    Path relativePath = Paths.get("../javaagent/build/libs").toAbsolutePath().normalize();
+    Path javaagentJar = findJavaagentJarInDirectory(relativePath);
 
-  // The sonatype repository can be very unreliable, so we retry a few times
-  private ResponseBody fetchBodyFrom(String url) throws IOException {
-    Request request = new Request.Builder().url(url).build();
-    IOException lastException = null;
+    if (javaagentJar != null) {
+      return javaagentJar;
+    }
 
-    for (int attempt = 0; attempt < 3; attempt++) {
-      try {
-        try (Response response = client.newCall(request).execute()) {
-          if (!response.isSuccessful()) {
-            throw new IOException("Unexpected HTTP code " + response.code() + " for " + url);
-          }
-          ResponseBody body = response.body();
-          if (body != null) {
-            byte[] data = body.bytes();
-            return ResponseBody.create(data, body.contentType());
-          } else {
-            throw new IOException("Response body is null");
-          }
-        }
-      } catch (IOException e) {
-        lastException = e;
-        if (attempt < 2) {
-          logger.warn("Attempt {} to fetch {} failed: {}. Retrying...", attempt + 1, url, e.getMessage());
+    // If not found, try from the project root (in case running from different location)
+    Path projectRoot = Paths.get(".").toAbsolutePath().normalize();
+    while (projectRoot.getParent() != null) {
+      Path gradlewFile = projectRoot.resolve("gradlew");
+      if (Files.exists(gradlewFile)) {
+        // Found the project root
+        Path javaagentLibsDir = projectRoot.resolve("javaagent/build/libs");
+        javaagentJar = findJavaagentJarInDirectory(javaagentLibsDir);
+        if (javaagentJar != null) {
+          return javaagentJar;
         }
+        break;
       }
+      projectRoot = projectRoot.getParent();
+    }
+
+    return null;
+  }
+
+  private Path findJavaagentJarInDirectory(Path directory) {
+    if (!Files.exists(directory) || !Files.isDirectory(directory)) {
+      return null;
+    }
+
+    try {
+      return Files.list(directory)
+          .filter(path -> {
+            String filename = path.getFileName().toString();
+            // Look for the main jar: opentelemetry-javaagent-VERSION.jar (no additional suffixes)
+            return filename.startsWith("opentelemetry-javaagent-") &&
+                   filename.endsWith(".jar") &&
+                   !filename.matches(".*-[a-z]+\\.jar"); // excludes anything with -word.jar pattern
+          })
+          .findFirst()
+          .orElse(null);
+    } catch (IOException e) {
+      logger.warn("Failed to list files in directory {}: {}", directory, e.getMessage());
+      return null;
     }
-    throw lastException;
   }
 }
 

conventions/src/main/kotlin/otel.nullaway-conventions.gradle.kts

@@ -18,19 +18,16 @@ nullaway {
 
 tasks {
   withType<JavaCompile>().configureEach {
-    if (name.contains("test", ignoreCase = true)) {
-      options.errorprone.nullaway {
-        enabled = false
-      }
-    } else {
-      options.errorprone.nullaway {
-        severity.set(CheckSeverity.ERROR)
-      }
-    }
     options.errorprone.nullaway {
+      if (name.contains("test", ignoreCase = true)) {
+        disable()
+      } else {
+        error()
+      }
       customInitializerAnnotations.add("org.openjdk.jmh.annotations.Setup")
       excludedFieldAnnotations.add("org.mockito.Mock")
       excludedFieldAnnotations.add("org.mockito.InjectMocks")
     }
   }
 }
+

instrumentation/kafka/kafka-clients/kafka-clients-0.11/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaclients/v0_11/ConsumerRecordsInstrumentation.java

@@ -19,10 +19,12 @@
 import io.opentelemetry.instrumentation.kafkaclients.common.v0_11.internal.TracingIterable;
 import io.opentelemetry.instrumentation.kafkaclients.common.v0_11.internal.TracingIterator;
 import io.opentelemetry.instrumentation.kafkaclients.common.v0_11.internal.TracingList;
+import io.opentelemetry.instrumentation.kafkaclients.common.v0_11.internal.TracingListIterator;
 import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation;
 import io.opentelemetry.javaagent.extension.instrumentation.TypeTransformer;
 import java.util.Iterator;
 import java.util.List;
+import java.util.ListIterator;
 import net.bytebuddy.asm.Advice;
 import net.bytebuddy.asm.Advice.AssignReturned;
 import net.bytebuddy.description.type.TypeDescription;
@@ -60,6 +62,13 @@ public void transform(TypeTransformer transformer) {
             .and(takesArguments(0))
             .and(returns(Iterator.class)),
         ConsumerRecordsInstrumentation.class.getName() + "$IteratorAdvice");
+    transformer.applyAdviceToMethod(
+        isMethod()
+            .and(isPublic())
+            .and(named("listIterator"))
+            .and(takesArguments(0))
+            .and(returns(ListIterator.class)),
+        ConsumerRecordsInstrumentation.class.getName() + "$ListIteratorAdvice");
   }
 
   @SuppressWarnings("unused")
@@ -118,4 +127,23 @@ public static <K, V> Iterator<ConsumerRecord<K, V>> wrap(
           iterator, consumerProcessInstrumenter(), wrappingEnabledSupplier(), consumerContext);
     }
   }
+
+  @SuppressWarnings("unused")
+  public static class ListIteratorAdvice {
+
+    @AssignReturned.ToReturned
+    @Advice.OnMethodExit(suppress = Throwable.class)
+    public static <K, V> ListIterator<ConsumerRecord<K, V>> wrap(
+        @Advice.This ConsumerRecords<?, ?> records,
+        @Advice.Return ListIterator<ConsumerRecord<K, V>> listIterator) {
+
+      // it's important not to suppress consumer span creation here because this instrumentation can
+      // leak the context and so there may be a leaked consumer span in the context, in which
+      // case it's important to overwrite the leaked span instead of suppressing the correct span
+      // (https://github.com/open-telemetry/opentelemetry-java-instrumentation/issues/1947)
+      KafkaConsumerContext consumerContext = KafkaConsumerContextUtil.get(records);
+      return TracingListIterator.wrap(
+          listIterator, consumerProcessInstrumenter(), wrappingEnabledSupplier(), consumerContext);
+    }
+  }
 }

instrumentation/kafka/kafka-clients/kafka-clients-0.11/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/kafkaclients/v0_11/KafkaClientDefaultTest.java

@@ -18,6 +18,7 @@
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.util.List;
+import java.util.ListIterator;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -155,8 +156,8 @@ void testPassThroughTombstone()
   }
 
   @DisplayName("test records(TopicPartition) kafka consume")
-  @Test
-  void testRecordsWithTopicPartitionKafkaConsume()
+  @ValueSource(booleans = {true, false})
+  void testRecordsWithTopicPartitionKafkaConsume(boolean testListIterator)
       throws ExecutionException, InterruptedException, TimeoutException {
     String greeting = "Hello from MockConsumer!";
     producer
@@ -172,9 +173,19 @@ void testRecordsWithTopicPartitionKafkaConsume()
     assertThat(recordsInPartition.size()).isEqualTo(1);
 
     // iterate over records to generate spans
-    for (ConsumerRecord<?, ?> record : recordsInPartition) {
-      assertThat(record.value()).isEqualTo(greeting);
-      assertThat(record.key()).isNull();
+    if (testListIterator) {
+      for (ListIterator<? extends ConsumerRecord<?, ?>> iterator =
+              recordsInPartition.listIterator();
+          iterator.hasNext(); ) {
+        ConsumerRecord<?, ?> record = iterator.next();
+        assertThat(record.value()).isEqualTo(greeting);
+        assertThat(record.key()).isNull();
+      }
+    } else {
+      for (ConsumerRecord<?, ?> record : recordsInPartition) {
+        assertThat(record.value()).isEqualTo(greeting);
+        assertThat(record.key()).isNull();
+      }
     }
 
     AtomicReference<SpanData> producerSpan = new AtomicReference<>();

instrumentation/kafka/kafka-clients/kafka-clients-common-0.11/library/src/main/java/io/opentelemetry/instrumentation/kafkaclients/common/v0_11/internal/TracingIterable.java

@@ -16,9 +16,9 @@
  */
 public class TracingIterable<K, V> implements Iterable<ConsumerRecord<K, V>> {
   private final Iterable<ConsumerRecord<K, V>> delegate;
-  private final Instrumenter<KafkaProcessRequest, Void> instrumenter;
-  private final BooleanSupplier wrappingEnabled;
-  private final KafkaConsumerContext consumerContext;
+  protected final Instrumenter<KafkaProcessRequest, Void> instrumenter;
+  protected final BooleanSupplier wrappingEnabled;
+  protected final KafkaConsumerContext consumerContext;
   private boolean firstIterator = true;
 
   protected TracingIterable(

instrumentation/kafka/kafka-clients/kafka-clients-common-0.11/library/src/main/java/io/opentelemetry/instrumentation/kafkaclients/common/v0_11/internal/TracingList.java

@@ -137,16 +137,14 @@ public int lastIndexOf(Object o) {
 
   @Override
   public ListIterator<ConsumerRecord<K, V>> listIterator() {
-    // TODO: the API for ListIterator is not really good to instrument it in context of Kafka
-    // Consumer so we will not do that for now
-    return delegate.listIterator();
+    return TracingListIterator.wrap(
+        delegate.listIterator(), instrumenter, wrappingEnabled, consumerContext);
   }
 
   @Override
   public ListIterator<ConsumerRecord<K, V>> listIterator(int index) {
-    // TODO: the API for ListIterator is not really good to instrument it in context of Kafka
-    // Consumer so we will not do that for now
-    return delegate.listIterator(index);
+    return TracingListIterator.wrap(
+        delegate.listIterator(index), instrumenter, wrappingEnabled, consumerContext);
   }
 
   @Override