Java 9 compatible ByteBuffer unmap operation (#7466)

The current bytebuffer unmap operations are not compatible with Java 9
and above. This change adapts the code from Apache Kafka to perform the
appropriate unmap operation based on JDK version at runtime.

see e554dc518e

LICENSE

@@ -266,6 +266,9 @@ SOURCE/JAVA-CORE
      copyright Netflix, Inc. (https://github.com/Netflix/spectator)
       * extensions-core/histogram/src/main/java/org/apache/druid/query/aggregation/histogram/FixedBucketsHistogram.java
 
+    This product contains ByteBuffer unmapping code adapted from Apache Kafka
+      * core/src/main/java/org/apache/druid/java/util/common/ByteBufferUtils.java
+
 
 MIT License
 ================================

core/src/main/java/org/apache/druid/java/util/common/ByteBufferUtils.java

@@ -19,16 +19,125 @@
 
 package org.apache.druid.java.util.common;
 
-import sun.misc.Cleaner;
-import sun.nio.ch.DirectBuffer;
+import org.apache.druid.utils.JvmUtils;
 
+import java.lang.invoke.MethodHandle;
+import java.lang.invoke.MethodHandles;
+import java.lang.invoke.MethodType;
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
 import java.nio.ByteBuffer;
 import java.nio.MappedByteBuffer;
+import java.util.Objects;
 
 /**
  */
 public class ByteBufferUtils
 {
+  // the following MethodHandle lookup code is adapted from Apache Kafka
+  // https://github.com/apache/kafka/blob/e554dc518eaaa0747899e708160275f95c4e525f/clients/src/main/java/org/apache/kafka/common/utils/MappedByteBuffers.java
+
+  // null if unmap is not supported
+  private static final MethodHandle UNMAP;
+
+  // null if unmap is supported
+  private static final RuntimeException UNMAP_NOT_SUPPORTED_EXCEPTION;
+
+  static {
+    Object unmap = null;
+    RuntimeException exception = null;
+    try {
+      unmap = lookupUnmapMethodHandle();
+    }
+    catch (RuntimeException e) {
+      exception = e;
+    }
+    if (unmap != null) {
+      UNMAP = (MethodHandle) unmap;
+      UNMAP_NOT_SUPPORTED_EXCEPTION = null;
+    } else {
+      UNMAP = null;
+      UNMAP_NOT_SUPPORTED_EXCEPTION = exception;
+    }
+  }
+
+  private static void clean(ByteBuffer buffer)
+  {
+    if (!buffer.isDirect()) {
+      throw new IllegalArgumentException("Unmapping only works with direct buffers");
+    }
+    if (UNMAP == null) {
+      throw new UnsupportedOperationException(UNMAP_NOT_SUPPORTED_EXCEPTION);
+    }
+
+    try {
+      UNMAP.invokeExact(buffer);
+    }
+    catch (Throwable throwable) {
+      throw new RuntimeException("Unable to unmap the mapped buffer", throwable);
+    }
+  }
+
+  private static MethodHandle lookupUnmapMethodHandle()
+  {
+    final MethodHandles.Lookup lookup = MethodHandles.lookup();
+    try {
+      if (JvmUtils.isIsJava9Compatible()) {
+        return unmapJava9(lookup);
+      } else {
+        return unmapJava7Or8(lookup);
+      }
+    }
+    catch (ReflectiveOperationException | RuntimeException e1) {
+      throw new UnsupportedOperationException("Unmapping is not supported on this platform, because internal " +
+                                              "Java APIs are not compatible with this Druid version", e1);
+    }
+  }
+
+  private static MethodHandle unmapJava7Or8(MethodHandles.Lookup lookup) throws ReflectiveOperationException
+  {
+    // "Compile" a MethodHandle that is roughly equivalent to the following lambda:
+    //
+    // (ByteBuffer buffer) -> {
+    //   sun.misc.Cleaner cleaner = ((java.nio.DirectByteBuffer) byteBuffer).cleaner();
+    //   if (nonNull(cleaner))
+    //     cleaner.clean();
+    //   else
+    //     noop(cleaner); // the noop is needed because MethodHandles#guardWithTest always needs both if and else
+    // }
+    //
+    Class<?> directBufferClass = Class.forName("java.nio.DirectByteBuffer");
+    Method m = directBufferClass.getMethod("cleaner");
+    m.setAccessible(true);
+    MethodHandle directBufferCleanerMethod = lookup.unreflect(m);
+    Class<?> cleanerClass = directBufferCleanerMethod.type().returnType();
+    MethodHandle cleanMethod = lookup.findVirtual(cleanerClass, "clean", MethodType.methodType(void.class));
+    MethodHandle nonNullTest = lookup.findStatic(Objects.class, "nonNull",
+                                                 MethodType.methodType(boolean.class, Object.class)
+    ).asType(MethodType.methodType(boolean.class, cleanerClass));
+    MethodHandle noop = MethodHandles.dropArguments(MethodHandles.constant(
+        Void.class,
+        null
+    ).asType(MethodType.methodType(void.class)), 0, cleanerClass);
+    MethodHandle unmapper = MethodHandles.filterReturnValue(
+        directBufferCleanerMethod,
+        MethodHandles.guardWithTest(nonNullTest, cleanMethod, noop)
+    ).asType(MethodType.methodType(void.class, ByteBuffer.class));
+    return unmapper;
+  }
+
+  private static MethodHandle unmapJava9(MethodHandles.Lookup lookup) throws ReflectiveOperationException
+  {
+    Class<?> unsafeClass = Class.forName("sun.misc.Unsafe");
+    MethodHandle unmapper = lookup.findVirtual(unsafeClass, "invokeCleaner",
+                                               MethodType.methodType(void.class, ByteBuffer.class)
+    );
+    Field f = unsafeClass.getDeclaredField("theUnsafe");
+    f.setAccessible(true);
+    Object theUnsafe = f.get(null);
+    return unmapper.bindTo(theUnsafe);
+  }
+
   /**
    * Releases memory held by the given direct ByteBuffer
    *
@@ -37,11 +146,10 @@ public class ByteBufferUtils
   public static void free(ByteBuffer buffer)
   {
     if (buffer.isDirect()) {
-      clean((DirectBuffer) buffer);
+      clean(buffer);
     }
   }
 
-
   /**
    * Un-maps the given memory mapped file
    *
@@ -51,12 +159,4 @@ public class ByteBufferUtils
   {
     free(buffer);
   }
-
-  private static void clean(DirectBuffer buffer)
-  {
-    final Cleaner cleaner = buffer.cleaner();
-    if (cleaner != null) {
-      cleaner.clean();
-    }
-  }
 }

core/src/main/java/org/apache/druid/utils/JvmUtils.java

@@ -23,9 +23,25 @@ import com.google.inject.Inject;
 
 import java.lang.management.ManagementFactory;
 import java.lang.management.ThreadMXBean;
+import java.util.StringTokenizer;
 
 public class JvmUtils
 {
+  private static final boolean IS_JAVA9_COMPATIBLE = isJava9Compatible(System.getProperty("java.specification.version"));
+
+  private static boolean isJava9Compatible(String versionString)
+  {
+    final StringTokenizer st = new StringTokenizer(versionString, ".");
+    int majorVersion = Integer.parseInt(st.nextToken());
+
+    return majorVersion >= 9;
+  }
+
+  public static boolean isIsJava9Compatible()
+  {
+    return IS_JAVA9_COMPATIBLE;
+  }
+
   @Inject
   private static RuntimeInfo runtimeInfo = new RuntimeInfo();