Merge branch 'main' of https://github.com/open-telemetry/opentelemetry-java into declarative-config-0.4

Author: jack-berg

.github/renovate.json5

@@ -17,6 +17,7 @@
     },
     {
       "matchPackageNames": [
+        "io.opentelemetry.contrib:opentelemetry-aws-xray-propagator",
         "io.opentelemetry.proto:opentelemetry-proto",
         "io.opentelemetry.semconv:opentelemetry-semconv-incubating"
       ],

dependencyManagement/build.gradle.kts

@@ -14,7 +14,7 @@ val DEPENDENCY_BOMS = listOf(
   // by FOSSA for containing EPL-licensed)
 
   "com.fasterxml.jackson:jackson-bom:2.18.3",
-  "com.google.guava:guava-bom:33.4.5-jre",
+  "com.google.guava:guava-bom:33.4.6-jre",
   "com.google.protobuf:protobuf-bom:4.29.3",
   "com.squareup.okhttp3:okhttp-bom:4.12.0",
   "com.squareup.okio:okio-bom:3.10.2", // applies to transitive dependencies of okhttp
@@ -83,7 +83,7 @@ val DEPENDENCIES = listOf(
   "eu.rekawek.toxiproxy:toxiproxy-java:2.1.7",
   "io.github.netmikey.logunit:logunit-jul:2.0.0",
   "io.jaegertracing:jaeger-client:1.8.1",
-  "io.opentelemetry.contrib:opentelemetry-aws-xray-propagator:1.39.0-alpha",
+  "io.opentelemetry.contrib:opentelemetry-aws-xray-propagator:1.45.0-alpha",
   "io.opentelemetry.semconv:opentelemetry-semconv-incubating:1.30.0-alpha",
   "io.opentelemetry.proto:opentelemetry-proto:1.5.0-alpha",
   "io.opentracing:opentracing-api:0.33.0",

sdk-extensions/incubator/build.gradle.kts

@@ -40,9 +40,7 @@ dependencies {
   testImplementation(project(":exporters:zipkin"))
   testImplementation(project(":sdk-extensions:jaeger-remote-sampler"))
   testImplementation(project(":extensions:trace-propagators"))
-  // As a part of the tests we check that we can parse examples without error. The https://github.com/open-telemetry/opentelemetry-configuration/blob/main/examples/kitchen-sink.yam contains a reference to the xray propagator
-  // TODO: add when updated to reflect new API locations
-  // testImplementation("io.opentelemetry.contrib:opentelemetry-aws-xray-propagator")
+  testImplementation("io.opentelemetry.contrib:opentelemetry-aws-xray-propagator")
   testImplementation("com.linecorp.armeria:armeria-junit5")
 
   testImplementation("com.google.guava:guava-testlib")

sdk-extensions/incubator/src/test/java/io/opentelemetry/sdk/extension/incubator/fileconfig/DeclarativeConfigurationCreateTest.java

@@ -94,10 +94,7 @@ void parseAndCreate_Examples(@TempDir Path tempDir)
                   "client_certificate_file: .*\n",
                   "client_certificate_file: "
                       + clientCertificatePath.replace("\\", "\\\\")
-                      + System.lineSeparator())
-              // TODO: remove once updated ComponentProvider SPI contract implemented in
-              // https://github.com/open-telemetry/opentelemetry-java-contrib/tree/main/aws-xray-propagator
-              .replaceAll("xray,", "");
+                      + System.lineSeparator());
       InputStream is =
           new ByteArrayInputStream(rewrittenExampleContent.getBytes(StandardCharsets.UTF_8));
 

sdk/trace-shaded-deps/src/main/java/io/opentelemetry/sdk/trace/internal/JcTools.java

@@ -64,21 +64,23 @@ public static long capacity(Queue<?> queue) {
    * @throws IllegalArgumentException if maxExportBatchSize is negative
    */
   @SuppressWarnings("unchecked")
-  public static <T> void drain(Queue<T> queue, int limit, Consumer<T> consumer) {
+  public static <T> int drain(Queue<T> queue, int limit, Consumer<T> consumer) {
     if (queue instanceof MessagePassingQueue) {
-      ((MessagePassingQueue<T>) queue).drain(consumer::accept, limit);
+      return ((MessagePassingQueue<T>) queue).drain(consumer::accept, limit);
     } else {
-      drainNonJcQueue(queue, limit, consumer);
+      return drainNonJcQueue(queue, limit, consumer);
     }
   }
 
-  private static <T> void drainNonJcQueue(
+  private static <T> int drainNonJcQueue(
       Queue<T> queue, int maxExportBatchSize, Consumer<T> consumer) {
     int polledCount = 0;
     T item;
-    while (polledCount++ < maxExportBatchSize && (item = queue.poll()) != null) {
+    while (polledCount < maxExportBatchSize && (item = queue.poll()) != null) {
       consumer.accept(item);
+      ++polledCount;
     }
+    return polledCount;
   }
 
   private JcTools() {}

sdk/trace/src/main/java/io/opentelemetry/sdk/trace/export/BatchSpanProcessor.java

@@ -173,6 +173,7 @@ private static final class Worker implements Runnable {
     private long nextExportTime;
 
     private final Queue<ReadableSpan> queue;
+    private final AtomicInteger queueSize = new AtomicInteger();
     // When waiting on the spans queue, exporter thread sets this atomic to the number of more
     // spans it needs before doing an export. Writer threads would then wait for the queue to reach
     // spansNeeded size before notifying the exporter thread about new entries.
@@ -237,7 +238,7 @@ private void addSpan(ReadableSpan span) {
       if (!queue.offer(span)) {
         processedSpansCounter.add(1, droppedAttrs);
       } else {
-        if (queue.size() >= spansNeeded.get()) {
+        if (queueSize.incrementAndGet() >= spansNeeded.get()) {
           signal.offer(true);
         }
       }
@@ -251,8 +252,7 @@ public void run() {
         if (flushRequested.get() != null) {
           flush();
         }
-        JcTools.drain(
-            queue, maxExportBatchSize - batch.size(), span -> batch.add(span.toSpanData()));
+        drain(maxExportBatchSize - batch.size());
 
         if (batch.size() >= maxExportBatchSize || System.nanoTime() >= nextExportTime) {
           exportCurrentBatch();
@@ -274,13 +274,17 @@ public void run() {
       }
     }
 
+    private int drain(int limit) {
+      int drained = JcTools.drain(queue, limit, span -> batch.add(span.toSpanData()));
+      queueSize.addAndGet(-drained);
+      return drained;
+    }
+
     private void flush() {
-      int spansToFlush = queue.size();
+      int spansToFlush = queueSize.get();
       while (spansToFlush > 0) {
-        ReadableSpan span = queue.poll();
-        assert span != null;
-        batch.add(span.toSpanData());
-        spansToFlush--;
+        int drained = drain(maxExportBatchSize - batch.size());
+        spansToFlush -= drained;
         if (batch.size() >= maxExportBatchSize) {
           exportCurrentBatch();
         }