Proactively avoid Unsafe on Java 23+

Author: trask

sdk/logs/build.gradle.kts

@@ -1,3 +1,5 @@
+import ru.vyarus.gradle.plugin.animalsniffer.AnimalSniffer
+
 plugins {
   id("otel.java-conventions")
   id("otel.publish-conventions")
@@ -9,10 +11,18 @@ plugins {
 description = "OpenTelemetry Log SDK"
 otelJava.moduleName.set("io.opentelemetry.sdk.logs")
 
+sourceSets {
+  main {
+    val traceShadedDeps = project(":sdk:trace-shaded-deps")
+    output.dir(traceShadedDeps.file("build/extracted/shadow"), "builtBy" to ":sdk:trace-shaded-deps:extractShadowJar")
+  }
+}
+
 dependencies {
   api(project(":api:all"))
   api(project(":sdk:common"))
   compileOnly(project(":api:incubator"))
+  compileOnly(project(":sdk:trace-shaded-deps"))
 
   annotationProcessor("com.google.auto.value:auto-value")
 

sdk/logs/src/main/java/io/opentelemetry/sdk/logs/export/BatchLogRecordProcessor.java

@@ -16,6 +16,7 @@
 import io.opentelemetry.sdk.logs.LogRecordProcessor;
 import io.opentelemetry.sdk.logs.ReadWriteLogRecord;
 import io.opentelemetry.sdk.logs.data.LogRecordData;
+import io.opentelemetry.sdk.trace.internal.JcTools;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -79,7 +80,7 @@ public static BatchLogRecordProcessorBuilder builder(LogRecordExporter logRecord
             scheduleDelayNanos,
             maxExportBatchSize,
             exporterTimeoutNanos,
-            new ArrayBlockingQueue<>(maxQueueSize)); // TODO: use JcTools.newFixedSizeQueue(..)
+            JcTools.newFixedSizeQueue(maxQueueSize));
     Thread workerThread = new DaemonThreadFactory(WORKER_THREAD_NAME).newThread(worker);
     workerThread.start();
   }

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

@@ -7,13 +7,13 @@
 
 import java.util.Objects;
 import java.util.Queue;
-import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Consumer;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 import org.jctools.queues.MessagePassingQueue;
 import org.jctools.queues.MpscArrayQueue;
+import org.jctools.queues.atomic.MpscAtomicArrayQueue;
 
 /**
  * Internal accessor of JCTools package for fast queues.
@@ -25,11 +25,15 @@ public final class JcTools {
 
   private static final AtomicBoolean queueCreationWarningLogged = new AtomicBoolean();
   private static final Logger logger = Logger.getLogger(JcTools.class.getName());
+  private static final boolean PROACTIVELY_AVOID_UNSAFE = proactivelyAvoidUnsafe();
 
   /**
    * Returns a new {@link Queue} appropriate for use with multiple producers and a single consumer.
    */
   public static <T> Queue<T> newFixedSizeQueue(int capacity) {
+    if (PROACTIVELY_AVOID_UNSAFE) {
+      return new MpscAtomicArrayQueue<>(capacity);
+    }
     try {
       return new MpscArrayQueue<>(capacity);
     } catch (java.lang.NoClassDefFoundError | java.lang.ExceptionInInitializerError e) {
@@ -41,7 +45,7 @@ public static <T> Queue<T> newFixedSizeQueue(int capacity) {
       }
       // Happens when modules such as jdk.unsupported are disabled in a custom JRE distribution,
       // or a security manager preventing access to Unsafe is installed.
-      return new ArrayBlockingQueue<>(capacity);
+      return new MpscAtomicArrayQueue<>(capacity);
     }
   }
 
@@ -50,11 +54,7 @@ public static <T> Queue<T> newFixedSizeQueue(int capacity) {
    * to use the shaded classes.
    */
   public static long capacity(Queue<?> queue) {
-    if (queue instanceof MessagePassingQueue) {
-      return ((MessagePassingQueue<?>) queue).capacity();
-    } else {
-      return (long) ((ArrayBlockingQueue<?>) queue).remainingCapacity() + queue.size();
-    }
+    return ((MessagePassingQueue<?>) queue).capacity();
   }
 
   /**
@@ -65,22 +65,26 @@ public static long capacity(Queue<?> queue) {
    */
   @SuppressWarnings("unchecked")
   public static <T> int drain(Queue<T> queue, int limit, Consumer<T> consumer) {
-    if (queue instanceof MessagePassingQueue) {
-      return ((MessagePassingQueue<T>) queue).drain(consumer::accept, limit);
-    } else {
-      return drainNonJcQueue(queue, limit, consumer);
-    }
+    return ((MessagePassingQueue<T>) queue).drain(consumer::accept, limit);
   }
 
-  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) {
-      consumer.accept(item);
-      ++polledCount;
+  private static boolean proactivelyAvoidUnsafe() {
+    double javaVersion = getJavaVersion();
+    // Avoid Unsafe on Java 23+ due to JEP-498 deprecation warnings:
+    // "WARNING: A terminally deprecated method in sun.misc.Unsafe has been called"
+    return javaVersion >= 23 || javaVersion == -1;
+  }
+
+  private static double getJavaVersion() {
+    String specVersion = System.getProperty("java.specification.version");
+    if (specVersion != null) {
+      try {
+        return Double.parseDouble(specVersion);
+      } catch (NumberFormatException exception) {
+        // ignore
+      }
     }
-    return polledCount;
+    return -1;
   }
 
   private JcTools() {}

sdk/trace-shaded-deps/src/test/java/io/opentelemetry/sdk/trace/internal/JcToolsSecurityManagerTest.java

@@ -11,7 +11,7 @@
 import java.security.AccessController;
 import java.security.PrivilegedAction;
 import java.util.Queue;
-import java.util.concurrent.ArrayBlockingQueue;
+import org.jctools.queues.atomic.MpscAtomicArrayQueue;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.condition.EnabledOnJre;
 import org.junit.jupiter.api.condition.JRE;
@@ -30,7 +30,7 @@ void newFixedSizeQueue_SunMiscProhibited() {
       Queue<Object> queue =
           AccessController.doPrivileged(
               (PrivilegedAction<Queue<Object>>) () -> JcTools.newFixedSizeQueue(10));
-      assertThat(queue).isInstanceOf(ArrayBlockingQueue.class);
+      assertThat(queue).isInstanceOf(MpscAtomicArrayQueue.class);
     } finally {
       System.setSecurityManager(null);
     }

sdk/trace-shaded-deps/src/test/java/io/opentelemetry/sdk/trace/internal/JcToolsTest.java

@@ -9,41 +9,19 @@
 
 import java.util.ArrayList;
 import java.util.Queue;
-import java.util.concurrent.ArrayBlockingQueue;
 import org.jctools.queues.MpscArrayQueue;
+import org.jctools.queues.atomic.MpscAtomicArrayQueue;
 import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.junit.jupiter.MockitoExtension;
-import org.mockito.junit.jupiter.MockitoSettings;
-import org.mockito.quality.Strictness;
 
-@ExtendWith(MockitoExtension.class)
-@MockitoSettings(strictness = Strictness.LENIENT)
 class JcToolsTest {
 
   ArrayList<String> batch = new ArrayList<>(10);
 
-  @Test
-  void drain_ArrayBlockingQueue() {
-    // Arrange
-    batch.add("Test3");
-    Queue<String> queue = new ArrayBlockingQueue<>(10);
-    queue.add("Test1");
-    queue.add("Test2");
-
-    // Act
-    JcTools.drain(queue, 5, batch::add);
-
-    // Assert
-    assertThat(batch).hasSize(3);
-    assertThat(queue).hasSize(0);
-  }
-
   @Test
   void drain_MessagePassingQueue() {
     // Arrange
     batch.add("Test3");
-    Queue<String> queue = new MpscArrayQueue<>(10);
+    Queue<String> queue = JcTools.newFixedSizeQueue(10);
     queue.add("Test1");
     queue.add("Test2");
 
@@ -58,7 +36,7 @@ void drain_MessagePassingQueue() {
   @Test
   void drain_MaxBatch() {
     // Arrange
-    Queue<String> queue = new MpscArrayQueue<>(10);
+    Queue<String> queue = JcTools.newFixedSizeQueue(10);
     queue.add("Test1");
     queue.add("Test2");
 
@@ -79,7 +57,10 @@ void newFixedSize_MpscQueue() {
     Queue<Object> objects = JcTools.newFixedSizeQueue(capacity);
 
     // Assert
-    assertThat(objects).isInstanceOf(MpscArrayQueue.class);
+    assertThat(objects)
+        .satisfiesAnyOf(
+            queue -> assertThat(queue).isInstanceOf(MpscArrayQueue.class),
+            queue -> assertThat(queue).isInstanceOf(MpscAtomicArrayQueue.class));
   }
 
   @Test
@@ -94,16 +75,4 @@ void capacity_MpscQueue() {
     // Assert
     assertThat(queueSize).isGreaterThan(capacity);
   }
-
-  @Test
-  void capacity_ArrayBlockingQueue() {
-    // Arrange
-    Queue<String> queue = new ArrayBlockingQueue<>(10);
-
-    // Act
-    long queueSize = JcTools.capacity(queue);
-
-    // Assert
-    assertThat(queueSize).isEqualTo(10);
-  }
 }

sdk/trace/build.gradle.kts

@@ -80,11 +80,3 @@ tasks {
     dependsOn(testing.suites)
   }
 }
-
-tasks {
-  withType<AnimalSniffer>().configureEach {
-    // We catch NoClassDefFoundError to fallback to non-jctools queues.
-    exclude("**/internal/shaded/jctools/**")
-    exclude("**/internal/JcTools*")
-  }
-}