Avoid race while retiring executors

Today, a race condition exists when retiring executors. Namely, if an
executor is retired and then the thread pool is terminated, the retiring
of the executor and the termination of the thread pool can race to
remove the retired executor from the queue of retired executors. More
precisely, when the executor is initially retired, it is placed on a
queue of retired executors, and then removed when it is successfully
shutdown. When the pool is terminated, it will also drain the queue of
retired executors. This leads to a time-of-check-time-of-use race where
the draining can see a retired executor on the queue but that retired
executor can be removed upon successful shutdown of that executor. This
leads to the draining attempting to remove an element from the queue
when there is none. This commit addresses this race condition by instead
safely polling the queue.

Relates #18333

core/src/main/java/org/elasticsearch/threadpool/ThreadPool.java

@@ -440,8 +440,11 @@ public class ThreadPool extends AbstractComponent implements Closeable {
                 ((ThreadPoolExecutor) executor.executor()).shutdownNow();
             }
         }
-        while (!retiredExecutors.isEmpty()) {
+
-            ((ThreadPoolExecutor) retiredExecutors.remove().executor()).shutdownNow();
+        ExecutorHolder holder;
+        while ((holder = retiredExecutors.poll()) != null) {
+            ThreadPoolExecutor executor = (ThreadPoolExecutor) holder.executor();
+            executor.shutdownNow();
         }
     }
 
@@ -452,10 +455,13 @@ public class ThreadPool extends AbstractComponent implements Closeable {
                 result &= ((ThreadPoolExecutor) executor.executor()).awaitTermination(timeout, unit);
             }
         }
-        while (!retiredExecutors.isEmpty()) {
+
-            ThreadPoolExecutor executor = (ThreadPoolExecutor) retiredExecutors.remove().executor();
+        ExecutorHolder holder;
+        while ((holder = retiredExecutors.poll()) != null) {
+            ThreadPoolExecutor executor = (ThreadPoolExecutor) holder.executor();
             result &= executor.awaitTermination(timeout, unit);
         }
+
         estimatedTimeThread.join(unit.toMillis(timeout));
         return result;
     }