HADOOP-19670 Replace Thread with SubjectPreservingThread to restore pre JDK22 Subject behaviour in Threads

Author: stoty

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/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/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) {

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

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

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

@@ -36,7 +36,7 @@
 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;
 import org.slf4j.LoggerFactory;
@@ -440,7 +440,7 @@ private void sendCallbackAndRemove(String caller,
   }
 
   @VisibleForTesting
-  final Thread watcherThread = new Thread(new Runnable() {
+  final Thread watcherThread = new SubjectInheritingThread(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,7 @@
 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;
 
@@ -930,7 +930,7 @@ private void executeAutoRenewalTask(final String userName,
                   new ThreadFactory() {
                     @Override
                     public Thread newThread(Runnable r) {
-                      Thread t = new Thread(r);
+                      Thread t = new SubjectInheritingThread(r);
                       t.setDaemon(true);
                       t.setName("TGT Renewer for " + userName);
                       return t;