Revert HADOOP-19668 and HADOOP-19670: Replace Thread with SubjectInheritingThread

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

@@ -22,7 +22,6 @@
 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;
@@ -106,16 +105,15 @@ public Collection<PropertyChange> getChangedProperties(
   /**
    * A background thread to apply configuration changes.
    */
-  private static class ReconfigurationThread extends SubjectInheritingThread {
+  private static class ReconfigurationThread extends Thread {
     private ReconfigurableBase parent;
 
     ReconfigurationThread(ReconfigurableBase base) {
-      super();
       this.parent = base;
     }
 
     // See {@link ReconfigurationServlet#applyChanges}
-    public void work() {
+    public void run() {
       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,7 +20,6 @@
 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;
 
@@ -108,7 +107,7 @@ void init() {
    */
   private void initRefreshThread(boolean runImmediately) {
     if (refreshInterval > 0) {
-      refreshUsed = new SubjectInheritingThread(new RefreshThread(this, runImmediately),
+      refreshUsed = new Thread(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,7 +30,6 @@
 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;
 
@@ -39,7 +38,7 @@
  */
 @InterfaceAudience.Private
 public class DelegationTokenRenewer
-    extends SubjectInheritingThread {
+    extends Thread {
   private static final Logger LOG = LoggerFactory
       .getLogger(DelegationTokenRenewer.class);
 
@@ -264,7 +263,7 @@ public <T extends FileSystem & Renewable> void removeRenewAction(
   }
 
   @Override
-  public void work() {
+  public void run() {
     for(;;) {
       RenewAction<?> action = null;
       try {

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

@@ -81,7 +81,6 @@
 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;
@@ -4092,7 +4091,7 @@ private interface StatisticsAggregator<T> {
     static {
       STATS_DATA_REF_QUEUE = new ReferenceQueue<>();
       // start a single daemon cleaner thread
-      STATS_DATA_CLEANER = new SubjectInheritingThread(new StatisticsDataReferenceCleaner());
+      STATS_DATA_CLEANER = new Thread(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 work() {
+    public void run() {
       while (shouldRun) {
         try { 
           loopUntilConnected();

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

@@ -17,7 +17,6 @@
  */
 package org.apache.hadoop.ha;
 
-import org.apache.hadoop.util.concurrent.SubjectInheritingThread;
 import org.slf4j.Logger;
 
 import java.io.BufferedReader;
@@ -51,7 +50,7 @@ enum StreamType {
     this.stream = stream;
     this.type = type;
 
-    thread = new SubjectInheritingThread(new Runnable() {
+    thread = new Thread(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 work() {
+            public void run() {
               for (; isRunning(this);) {
                 final long waitTime = checkCalls();
                 tryStop(this);

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

@@ -54,7 +54,6 @@
 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;
@@ -408,7 +407,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 SubjectInheritingThread {
+  private class Connection extends Thread {
     private InetSocketAddress server;             // server ip:port
     private final ConnectionId remoteId;          // connection id
     private AuthMethod authMethod; // authentication method
@@ -449,7 +448,7 @@ private class Connection extends SubjectInheritingThread {
         Consumer<Connection> removeMethod) {
       this.remoteId = remoteId;
       this.server = remoteId.getAddress();
-      this.rpcRequestThread = new SubjectInheritingThread(new RpcRequestSender(),
+      this.rpcRequestThread = new Thread(new RpcRequestSender(),
           "IPC Parameter Sending Thread for " + remoteId);
       this.rpcRequestThread.setDaemon(true);
 
@@ -1127,7 +1126,7 @@ private synchronized void sendPing() throws IOException {
     }
 
     @Override
-    public void work() {
+    public void run() {
       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

@@ -68,6 +68,7 @@
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.LongAdder;
@@ -81,6 +82,7 @@
 import org.apache.hadoop.classification.InterfaceAudience.Private;
 import org.apache.hadoop.classification.InterfaceAudience.Public;
 import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.classification.VisibleForTesting;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.conf.Configuration.IntegerRanges;
 import org.apache.hadoop.fs.CommonConfigurationKeys;
@@ -106,6 +108,7 @@
 import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RPCTraceInfoProto;
 import org.apache.hadoop.net.NetUtils;
 import org.apache.hadoop.security.AccessControlException;
+import org.apache.hadoop.security.AuthorizationContext;
 import org.apache.hadoop.security.SaslMechanismFactory;
 import org.apache.hadoop.security.SaslPropertiesResolver;
 import org.apache.hadoop.security.SaslRpcServer;
@@ -124,24 +127,19 @@
 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;
 import org.apache.hadoop.tracing.TraceScope;
 import org.apache.hadoop.tracing.Tracer;
 import org.apache.hadoop.tracing.TraceUtils;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import org.apache.hadoop.classification.VisibleForTesting;
-
 import org.apache.hadoop.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
 import org.apache.hadoop.thirdparty.protobuf.ByteString;
 import org.apache.hadoop.thirdparty.protobuf.CodedOutputStream;
 import org.apache.hadoop.thirdparty.protobuf.Message;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.apache.hadoop.security.AuthorizationContext;
 
 /** An abstract IPC service.  IPC calls take a single {@link Writable} as a
  * parameter, and return a {@link Writable} as their value.  A service runs on
@@ -1473,7 +1471,7 @@ public String toString() {
   }
 
   /** Listens on the socket. Creates jobs for the handler threads*/
-  private class Listener extends SubjectInheritingThread {
+  private class Listener extends Thread {
 
     private ServerSocketChannel acceptChannel = null; //the accept channel
     private Selector selector = null; //the selector that we use for the server
@@ -1522,7 +1520,7 @@ void setIsAuxiliary() {
       this.isOnAuxiliaryPort = true;
     }
 
-    private class Reader extends SubjectInheritingThread {
+    private class Reader extends Thread {
       final private BlockingQueue<Connection> pendingConnections;
       private final Selector readSelector;
 
@@ -1535,7 +1533,7 @@ private class Reader extends SubjectInheritingThread {
       }
 
       @Override
-      public void work() {
+      public void run() {
         LOG.info("Starting " + Thread.currentThread().getName());
         try {
           doRunLoop();
@@ -1614,7 +1612,7 @@ void shutdown() {
     }
 
     @Override
-    public void work() {
+    public void run() {
       LOG.info(Thread.currentThread().getName() + ": starting");
       SERVER.set(Server.this);
       connectionManager.startIdleScan();
@@ -1762,7 +1760,7 @@ Reader getReader() {
   }
 
   // Sends responses of RPC back to clients.
-  private class Responder extends SubjectInheritingThread {
+  private class Responder extends Thread {
     private final Selector writeSelector;
     private int pending;         // connections waiting to register
 
@@ -1774,7 +1772,7 @@ private class Responder extends SubjectInheritingThread {
     }
 
     @Override
-    public void work() {
+    public void run() {
       LOG.info(Thread.currentThread().getName() + ": starting");
       SERVER.set(Server.this);
       try {
@@ -3240,15 +3238,15 @@ private void internalQueueCall(Call call, boolean blocking)
   }
 
   /** Handles queued calls . */
-  private class Handler extends SubjectInheritingThread {
+  private class Handler extends Thread {
     public Handler(int instanceNumber) {
       this.setDaemon(true);
       this.setName("IPC Server handler "+ instanceNumber +
           " on default port " + port);
     }
 
     @Override
-    public void work() {
+    public void run() {
       LOG.debug("{}: starting", Thread.currentThread().getName());
       SERVER.set(Server.this);
       while (running) {

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/impl/MetricsSinkAdapter.java

@@ -34,7 +34,6 @@
 import org.apache.hadoop.metrics2.MetricsFilter;
 import org.apache.hadoop.metrics2.MetricsSink;
 import org.apache.hadoop.util.Time;
-import org.apache.hadoop.util.concurrent.SubjectInheritingThread;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -49,7 +48,7 @@ class MetricsSinkAdapter implements SinkQueue.Consumer<MetricsBuffer> {
   private final MetricsSink sink;
   private final MetricsFilter sourceFilter, recordFilter, metricFilter;
   private final SinkQueue<MetricsBuffer> queue;
-  private final SubjectInheritingThread sinkThread;
+  private final Thread sinkThread;
   private volatile boolean stopping = false;
   private volatile boolean inError = false;
   private final int periodMs, firstRetryDelay, retryCount;
@@ -85,8 +84,8 @@ class MetricsSinkAdapter implements SinkQueue.Consumer<MetricsBuffer> {
                                   "Dropped updates per sink", 0);
     qsize = registry.newGauge("Sink_"+ name + "Qsize", "Queue size", 0);
 
-    sinkThread = new SubjectInheritingThread() {
-      @Override public void work() {
+    sinkThread = new Thread() {
+      @Override public void run() {
         publishMetricsFromQueue();
       }
     };

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/net/unix/DomainSocketWatcher.java

@@ -36,7 +36,6 @@
 import org.apache.hadoop.util.NativeCodeLoader;
 import org.apache.hadoop.classification.VisibleForTesting;
 import org.apache.hadoop.util.Preconditions;
-import org.apache.hadoop.util.concurrent.SubjectInheritingThread;
 
 import org.apache.hadoop.thirdparty.com.google.common.util.concurrent.Uninterruptibles;
 import org.slf4j.Logger;
@@ -441,7 +440,7 @@ private void sendCallbackAndRemove(String caller,
   }
 
   @VisibleForTesting
-  final Thread watcherThread = new SubjectInheritingThread(new Runnable() {
+  final Thread watcherThread = new Thread(new Runnable() {
     @Override
     public void run() {
       if (LOG.isDebugEnabled()) {

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/UserGroupInformation.java

@@ -89,7 +89,6 @@
 import org.apache.hadoop.util.Shell;
 import org.apache.hadoop.util.StringUtils;
 import org.apache.hadoop.util.Time;
-import org.apache.hadoop.util.concurrent.SubjectInheritingThread;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -931,7 +930,7 @@ private void executeAutoRenewalTask(final String userName,
                   new ThreadFactory() {
                     @Override
                     public Thread newThread(Runnable r) {
-                      Thread t = new SubjectInheritingThread(r);
+                      Thread t = new Thread(r);
                       t.setDaemon(true);
                       t.setName("TGT Renewer for " + userName);
                       return t;

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/token/delegation/AbstractDelegationTokenSecretManager.java

@@ -60,9 +60,9 @@
 import org.apache.hadoop.security.token.Token;
 import org.apache.hadoop.util.Daemon;
 import org.apache.hadoop.util.Time;
-import org.apache.hadoop.util.concurrent.SubjectInheritingThread;
 import org.apache.hadoop.util.Preconditions;
 import org.apache.hadoop.util.functional.InvocationRaisingIOE;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -912,12 +912,12 @@ public boolean isRunning() {
     return running;
   }
 
-  private class ExpiredTokenRemover extends SubjectInheritingThread {
+  private class ExpiredTokenRemover extends Thread {
     private long lastMasterKeyUpdate;
     private long lastTokenCacheCleanup;
 
     @Override
-    public void work() {
+    public void run() {
       LOG.info("Starting expired delegation token remover thread, "
           + "tokenRemoverScanInterval=" + tokenRemoverScanInterval
           / (60 * 1000) + " min(s)");

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/service/launcher/InterruptEscalator.java

@@ -24,7 +24,6 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.apache.hadoop.util.Preconditions;
-import org.apache.hadoop.util.concurrent.SubjectInheritingThread;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -117,7 +116,7 @@ public void interrupted(IrqHandler.InterruptData interruptData) {
       //start an async shutdown thread with a timeout
       ServiceForcedShutdown shutdown =
           new ServiceForcedShutdown(service, shutdownTimeMillis);
-      Thread thread = new SubjectInheritingThread(shutdown);
+      Thread thread = new Thread(shutdown);
       thread.setDaemon(true);
       thread.setName("Service Forced Shutdown");
       thread.start();