attempt to fix the Subject propagation nonsense

Author: stoty

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

@@ -18,6 +18,7 @@
 public class SubjectUtil {
 	private static final MethodHandle CALL_AS = lookupCallAs();
 	private static final MethodHandle CURRENT = lookupCurrent();
+	private static boolean HAS_CALL_AS = true;
 
 	private SubjectUtil() {
 	}
@@ -33,6 +34,7 @@ private static MethodHandle lookupCallAs() {
 				return lookup.findStatic(Subject.class, "callAs",
 						MethodType.methodType(Object.class, Subject.class, Callable.class));
 			} catch (NoSuchMethodException x) {
+			  HAS_CALL_AS = false;
 				try {
 					// Lookup the old API.
 					MethodType oldSignature = MethodType.methodType(Object.class, Subject.class,
@@ -168,6 +170,25 @@ public static Subject current() {
 		}
 	}
 
+  public static Runnable wrap(Runnable r) {
+    if (!HAS_CALL_AS) {
+      return r;
+    }
+    Subject s = current();
+    return () -> callAs(s, () -> {
+      r.run();
+      return null;
+    });
+  }
+
+  public static <T> Callable<T> wrap(Callable<T> c) {
+    if (!HAS_CALL_AS) {
+      return c;
+    }
+    Subject s = current();
+    return () -> callAs(s, () -> c.call());
+  }
+
 	@SuppressWarnings("unused")
 	private static <T> PrivilegedExceptionAction<T> callableToPrivilegedExceptionAction(Callable<T> callable) {
 		return callable::call;

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/concurrent/HadoopScheduledThreadPoolExecutor.java

@@ -20,12 +20,18 @@
 
 package org.apache.hadoop.util.concurrent;
 
+import org.apache.hadoop.util.SubjectUtil;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.concurrent.Callable;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.RejectedExecutionHandler;
+import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
 
 /** An extension of ScheduledThreadPoolExecutor that provides additional
  * functionality. */
@@ -68,4 +74,53 @@ protected void afterExecute(Runnable r, Throwable t) {
     super.afterExecute(r, t);
     ExecutorHelper.logThrowableFromAfterExecute(r, t);
   }
+
+  /**
+   * @throws RejectedExecutionException {@inheritDoc}
+   * @throws NullPointerException       {@inheritDoc}
+   */
+  @Override
+  public ScheduledFuture<?> schedule(Runnable command,
+                                     long delay,
+                                     TimeUnit unit) {
+      return super.schedule(SubjectUtil.wrap(command), delay, unit);
+  }
+
+  /**
+   * @throws RejectedExecutionException {@inheritDoc}
+   * @throws NullPointerException       {@inheritDoc}
+   */
+  @Override
+  public <V> ScheduledFuture<V> schedule(Callable<V> callable,
+                                         long delay,
+                                         TimeUnit unit) {
+    return super.schedule(SubjectUtil.wrap(callable), delay, unit);
+  }
+
+  /**
+   * @throws RejectedExecutionException {@inheritDoc}
+   * @throws NullPointerException       {@inheritDoc}
+   * @throws IllegalArgumentException   {@inheritDoc}
+   */
+  @Override
+  public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
+                                                long initialDelay,
+                                                long period,
+                                                TimeUnit unit) {
+    return super.scheduleAtFixedRate(SubjectUtil.wrap(command), initialDelay, period, unit);
+  }
+
+  /**
+   * @throws RejectedExecutionException {@inheritDoc}
+   * @throws NullPointerException       {@inheritDoc}
+   * @throws IllegalArgumentException   {@inheritDoc}
+   */
+  @Override
+  public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
+                                                   long initialDelay,
+                                                   long delay,
+                                                   TimeUnit unit) {
+    return super.scheduleWithFixedDelay(SubjectUtil.wrap(command), initialDelay, delay, unit);
+  }
+
 }

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/concurrent/HadoopThreadPoolExecutor.java

@@ -20,6 +20,7 @@
 
 package org.apache.hadoop.util.concurrent;
 
+import org.apache.hadoop.util.SubjectUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -76,6 +77,11 @@ public HadoopThreadPoolExecutor(int corePoolSize,
         threadFactory, handler);
   }
 
+  @Override
+  public void execute(Runnable command) {
+    super.execute(SubjectUtil.wrap(command));
+  }
+
   @Override
   protected void beforeExecute(Thread t, Runnable r) {
     if (LOG.isDebugEnabled()) {