HADOOP-19574. Restore Subject propagation semantics for Java 22+

hadoop-common-project/hadoop-auth/src/main/java/org/apache/hadoop/security/authentication/util/SubjectUtil.java

@@ -56,6 +56,18 @@ public final class SubjectUtil {
       HAS_CALL_AS ? null : lookupDoAsThrowException();
   private static final MethodHandle CURRENT = lookupCurrent();
 
+  // copied from org.apache.hadoop.util.Shell to break circular dependency
+  // "1.8"->8, "9"->9, "10"->10
+  private static final int JAVA_SPEC_VER = Math.max(8,
+      Integer.parseInt(System.getProperty("java.specification.version").split("\\.")[0]));
+
+  public static final boolean THREAD_INHERITS_SUBJECT = checkThreadInheritsSubject();
+
+  /**
+   * Try to return the method handle for Subject#callAs()
+   *
+   * @return the method handle, or null if the Java version does not have it
+   */
   private static MethodHandle lookupCallAs() {
     MethodHandles.Lookup lookup = MethodHandles.lookup();
     try {
@@ -71,6 +83,38 @@ private static MethodHandle lookupCallAs() {
     }
   }
 
+  /**
+   * Determine whether we need to explicitly propagate the Subject into new Threads.
+   *
+   * @return true if new Threads inherit the Subject from the parent
+   */
+  private static boolean checkThreadInheritsSubject() {
+
+    boolean securityManagerEnabled = true;
+    try {
+      // TODO this needs SecurityManager to compile, use reflection to look it up instead
+      SecurityManager sm = System.getSecurityManager();
+      System.setSecurityManager(sm);
+    } catch (UnsupportedOperationException e) {
+      // JDK24+ unconditionally throws this, so we don't need to check for JDK24+
+      // explicitly
+      securityManagerEnabled = false;
+    } catch (Throwable t) {
+      // don't care
+    }
+
+    return JAVA_SPEC_VER < 22 || securityManagerEnabled;
+  }
+
+  /**
+   * Look up the method handle for Subject#doAs(PrivilegedAction)
+   *
+   * This is only called if Subject#callAs() does not exist.
+   * If we can't fall back to doAs(), that's a hard error.
+   *
+   * @return the method handle
+   * @throws ExceptionInInitializerError if unable to get the method handle
+   */
   private static MethodHandle lookupDoAs() {
     MethodHandles.Lookup lookup = MethodHandles.lookup();
     try {
@@ -82,6 +126,15 @@ private static MethodHandle lookupDoAs() {
     }
   }
 
+  /**
+   * Look up the method handle for Subject#doAs(PrivilegedExceptionAction)
+   *
+   * This is only called if Subject#callAs() does not exist.
+   * If we can't fall back to doAs(), that's a hard error.
+   *
+   * @return the method handle
+   * @throws ExceptionInInitializerError if unable to get the method handle
+   */
   private static MethodHandle lookupDoAsThrowException() {
     MethodHandles.Lookup lookup = MethodHandles.lookup();
     try {
@@ -93,6 +146,15 @@ private static MethodHandle lookupDoAsThrowException() {
     }
   }
 
+  /**
+   * Look up the method handle for Subject#current().
+   *
+   * If Subject#current() is not present, fall back to returning
+   * a method handle for Subject.getSubject(AccessController.getContext())
+   *
+   * @return the method handle or null if it does not exist
+   * @throws ExceptionInInitializerError if neither current() nor the fallback is found
+   */
   private static MethodHandle lookupCurrent() {
     MethodHandles.Lookup lookup = MethodHandles.lookup();
     try {
@@ -112,6 +174,15 @@ private static MethodHandle lookupCurrent() {
     }
   }
 
+  /**
+   * Look up the method handle for Subject#getSubject(AccessControlContext)
+   *
+   * This is only called if Subject#current() does not exist.
+   * If we can't fall back to getSubject(), that's a hard error.
+   *
+   * @return the method handle
+   * @throws ExceptionInInitializerError if cannot get the handle
+   */
   private static MethodHandle lookupGetSubject() {
     MethodHandles.Lookup lookup = MethodHandles.lookup();
     try {
@@ -124,6 +195,15 @@ private static MethodHandle lookupGetSubject() {
     }
   }
 
+  /**
+   * Look up the method handle for AccessController.getAccessControlContext()
+   *
+   * This is only called if Subject#current() does not exist.
+   * If we can't find this method, then we can't fall back which is hard error.
+   *
+   * @return the method handle
+   * @throws ExceptionInInitializerError if cannot get the handle
+   */
   private static MethodHandle lookupGetContext() {
     try {
       // Use reflection to work with Java versions that have and don't have
@@ -264,6 +344,13 @@ public static Subject current() {
     }
   }
 
+  /**
+   * Convert a Callable into a PrivilegedAction
+   *
+   * @param <T> return type
+   * @param callable to be converted
+   * @return PrivilegedAction wrapping the callable
+   */
   private static <T> PrivilegedAction<T> callableToPrivilegedAction(
       Callable<T> callable) {
     return () -> {
@@ -275,11 +362,25 @@ private static <T> PrivilegedAction<T> callableToPrivilegedAction(
     };
   }
 
+  /**
+   * Convert a PrivilegedExceptionAction into a Callable
+   *
+   * @param <T> return type
+   * @param action to be wrapped
+   * @return Callable wrapping the action
+   */
   private static <T> Callable<T> privilegedExceptionActionToCallable(
       PrivilegedExceptionAction<T> action) {
     return action::run;
   }
 
+  /**
+   * Convert a PrivilegedAction into a Callable
+   *
+   * @param <T> return type
+   * @param action to be wrapped
+   * @return Callable wrapping the action
+   */
   private static <T> Callable<T> privilegedActionToCallable(
       PrivilegedAction<T> action) {
     return action::run;

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/conf/ReconfigurableBase.java

@@ -22,6 +22,7 @@
 import org.apache.hadoop.util.Preconditions;
 import org.apache.hadoop.thirdparty.com.google.common.collect.Maps;
 import org.apache.hadoop.util.Time;
+import org.apache.hadoop.util.concurrent.SubjectInheritingThread;
 import org.apache.hadoop.conf.ReconfigurationUtil.PropertyChange;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -105,15 +106,16 @@ public Collection<PropertyChange> getChangedProperties(
   /**
    * A background thread to apply configuration changes.
    */
-  private static class ReconfigurationThread extends Thread {
+  private static class ReconfigurationThread extends SubjectInheritingThread {
     private ReconfigurableBase parent;
 
     ReconfigurationThread(ReconfigurableBase base) {
+      super();
       this.parent = base;
     }
 
     // See {@link ReconfigurationServlet#applyChanges}
-    public void run() {
+    public void work() {
       LOG.info("Starting reconfiguration task.");
       final Configuration oldConf = parent.getConf();
       final Configuration newConf = parent.getNewConf();

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/CachingGetSpaceUsed.java

@@ -20,6 +20,7 @@
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.classification.VisibleForTesting;
+import org.apache.hadoop.util.concurrent.SubjectInheritingThread;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -107,7 +108,7 @@ void init() {
    */
   private void initRefreshThread(boolean runImmediately) {
     if (refreshInterval > 0) {
-      refreshUsed = new Thread(new RefreshThread(this, runImmediately),
+      refreshUsed = new SubjectInheritingThread(new RefreshThread(this, runImmediately),
           "refreshUsed-" + dirPath);
       refreshUsed.setDaemon(true);
       refreshUsed.start();

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/DelegationTokenRenewer.java

@@ -30,6 +30,7 @@
 import org.apache.hadoop.security.token.Token;
 import org.apache.hadoop.security.token.TokenIdentifier;
 import org.apache.hadoop.util.Time;
+import org.apache.hadoop.util.concurrent.SubjectInheritingThread;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -38,7 +39,7 @@
  */
 @InterfaceAudience.Private
 public class DelegationTokenRenewer
-    extends Thread {
+    extends SubjectInheritingThread {
   private static final Logger LOG = LoggerFactory
       .getLogger(DelegationTokenRenewer.class);
 
@@ -263,7 +264,7 @@ public <T extends FileSystem & Renewable> void removeRenewAction(
   }
 
   @Override
-  public void run() {
+  public void work() {
     for(;;) {
       RenewAction<?> action = null;
       try {

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileSystem.java

@@ -81,6 +81,7 @@
 import org.apache.hadoop.util.ReflectionUtils;
 import org.apache.hadoop.util.ShutdownHookManager;
 import org.apache.hadoop.util.StringUtils;
+import org.apache.hadoop.util.concurrent.SubjectInheritingThread;
 import org.apache.hadoop.tracing.Tracer;
 import org.apache.hadoop.tracing.TraceScope;
 import org.apache.hadoop.util.Preconditions;
@@ -4087,7 +4088,7 @@ private interface StatisticsAggregator<T> {
     static {
       STATS_DATA_REF_QUEUE = new ReferenceQueue<>();
       // start a single daemon cleaner thread
-      STATS_DATA_CLEANER = new Thread(new StatisticsDataReferenceCleaner());
+      STATS_DATA_CLEANER = new SubjectInheritingThread(new StatisticsDataReferenceCleaner());
       STATS_DATA_CLEANER.
           setName(StatisticsDataReferenceCleaner.class.getName());
       STATS_DATA_CLEANER.setDaemon(true);

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ha/HealthMonitor.java

@@ -283,7 +283,7 @@ public void uncaughtException(Thread t, Throwable e) {
     }
 
     @Override
-    public void run() {
+    public void work() {
       while (shouldRun) {
         try { 
           loopUntilConnected();

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ha/StreamPumper.java

@@ -17,6 +17,7 @@
  */
 package org.apache.hadoop.ha;
 
+import org.apache.hadoop.util.concurrent.SubjectInheritingThread;
 import org.slf4j.Logger;
 
 import java.io.BufferedReader;
@@ -50,7 +51,7 @@ enum StreamType {
     this.stream = stream;
     this.type = type;
 
-    thread = new Thread(new Runnable() {
+    thread = new SubjectInheritingThread(new Runnable() {
       @Override
       public void run() {
         try {

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/retry/AsyncCallHandler.java

@@ -158,7 +158,7 @@ void tryStart() {
         if (running.compareAndSet(null, current)) {
           final Daemon daemon = new Daemon() {
             @Override
-            public void run() {
+            public void work() {
               for (; isRunning(this);) {
                 final long waitTime = checkCalls();
                 tryStop(this);

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Client.java

@@ -54,6 +54,7 @@
 import org.apache.hadoop.util.StringUtils;
 import org.apache.hadoop.util.Time;
 import org.apache.hadoop.util.concurrent.AsyncGet;
+import org.apache.hadoop.util.concurrent.SubjectInheritingThread;
 import org.apache.hadoop.tracing.Span;
 import org.apache.hadoop.tracing.Tracer;
 import org.slf4j.Logger;
@@ -407,7 +408,7 @@ public synchronized void setRpcResponse(Writable rpcResponse) {
   /** Thread that reads responses and notifies callers.  Each connection owns a
    * socket connected to a remote address.  Calls are multiplexed through this
    * socket: responses may be delivered out of order. */
-  private class Connection extends Thread {
+  private class Connection extends SubjectInheritingThread {
     private InetSocketAddress server;             // server ip:port
     private final ConnectionId remoteId;          // connection id
     private AuthMethod authMethod; // authentication method
@@ -448,7 +449,7 @@ private class Connection extends Thread {
         Consumer<Connection> removeMethod) {
       this.remoteId = remoteId;
       this.server = remoteId.getAddress();
-      this.rpcRequestThread = new Thread(new RpcRequestSender(),
+      this.rpcRequestThread = new SubjectInheritingThread(new RpcRequestSender(),
           "IPC Parameter Sending Thread for " + remoteId);
       this.rpcRequestThread.setDaemon(true);
 
@@ -1126,7 +1127,7 @@ private synchronized void sendPing() throws IOException {
     }
 
     @Override
-    public void run() {
+    public void work() {
       try {
         // Don't start the ipc parameter sending thread until we start this
         // thread, because the shutdown logic only gets triggered if this

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java

@@ -124,6 +124,8 @@
 import org.apache.hadoop.util.ProtoUtil;
 import org.apache.hadoop.util.StringUtils;
 import org.apache.hadoop.util.Time;
+import org.apache.hadoop.util.concurrent.SubjectInheritingThread;
+
 import java.util.concurrent.atomic.AtomicBoolean;
 import org.apache.hadoop.tracing.Span;
 import org.apache.hadoop.tracing.SpanContext;
@@ -1471,7 +1473,7 @@ public String toString() {
   }
 
   /** Listens on the socket. Creates jobs for the handler threads*/
-  private class Listener extends Thread {
+  private class Listener extends SubjectInheritingThread {
 
     private ServerSocketChannel acceptChannel = null; //the accept channel
     private Selector selector = null; //the selector that we use for the server
@@ -1520,7 +1522,7 @@ void setIsAuxiliary() {
       this.isOnAuxiliaryPort = true;
     }
 
-    private class Reader extends Thread {
+    private class Reader extends SubjectInheritingThread {
       final private BlockingQueue<Connection> pendingConnections;
       private final Selector readSelector;
 
@@ -1533,7 +1535,7 @@ private class Reader extends Thread {
       }
 
       @Override
-      public void run() {
+      public void work() {
         LOG.info("Starting " + Thread.currentThread().getName());
         try {
           doRunLoop();
@@ -1612,7 +1614,7 @@ void shutdown() {
     }
 
     @Override
-    public void run() {
+    public void work() {
       LOG.info(Thread.currentThread().getName() + ": starting");
       SERVER.set(Server.this);
       connectionManager.startIdleScan();
@@ -1760,7 +1762,7 @@ Reader getReader() {
   }
 
   // Sends responses of RPC back to clients.
-  private class Responder extends Thread {
+  private class Responder extends SubjectInheritingThread {
     private final Selector writeSelector;
     private int pending;         // connections waiting to register
 
@@ -1772,7 +1774,7 @@ private class Responder extends Thread {
     }
 
     @Override
-    public void run() {
+    public void work() {
       LOG.info(Thread.currentThread().getName() + ": starting");
       SERVER.set(Server.this);
       try {
@@ -3219,15 +3221,15 @@ private void internalQueueCall(Call call, boolean blocking)
   }
 
   /** Handles queued calls . */
-  private class Handler extends Thread {
+  private class Handler extends SubjectInheritingThread {
     public Handler(int instanceNumber) {
       this.setDaemon(true);
       this.setName("IPC Server handler "+ instanceNumber +
           " on default port " + port);
     }
 
     @Override
-    public void run() {
+    public void work() {
       LOG.debug("{}: starting", Thread.currentThread().getName());
       SERVER.set(Server.this);
       while (running) {