Fix thread context propagation in TransportClusterUpdateSettingsAction

This fixes the settings API to properly return deprecation warning
headers when updating deprecated settings, while properly associating
them with the correct correlation id.

Previously, the correlation id of the original context wasn't propagated
to the cluster state update task.
However, using `newRestorableContext(preserveResponseHeaders=true)`
isn't enough to also - in case - propagate warning headers back to
the task's thread context and hence can't be captured by the task
executor in MasterService to return the response headers back to the
caller.

This PR adds a new variant of `newRestorableContext` that supports
propagation of response headers if required.

Fixes #108628 (DeprecationHttpIT.testDeprecatedSettingsReturnWarnings)

Author: mosche

muted-tests.yml

@@ -66,9 +66,6 @@ tests:
 - class: org.elasticsearch.xpack.test.rest.XPackRestIT
   method: test {p0=transform/transforms_start_stop/Test start already started transform}
   issue: https://github.com/elastic/elasticsearch/issues/98802
-- class: org.elasticsearch.xpack.deprecation.DeprecationHttpIT
-  method: testDeprecatedSettingsReturnWarnings
-  issue: https://github.com/elastic/elasticsearch/issues/108628
 - class: org.elasticsearch.xpack.shutdown.NodeShutdownIT
   method: testAllocationPreventedForRemoval
   issue: https://github.com/elastic/elasticsearch/issues/116363

server/src/main/java/org/elasticsearch/action/admin/cluster/settings/TransportClusterUpdateSettingsAction.java

@@ -30,6 +30,7 @@
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.settings.SettingsUpdater;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
+import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.core.SuppressForbidden;
 import org.elasticsearch.injection.guice.Inject;
 import org.elasticsearch.reservedstate.action.ReservedClusterSettingsAction;
@@ -40,6 +41,7 @@
 import java.util.HashSet;
 import java.util.Optional;
 import java.util.Set;
+import java.util.function.Supplier;
 
 import static org.elasticsearch.common.settings.AbstractScopedSettings.ARCHIVED_SETTINGS_PREFIX;
 
@@ -155,7 +157,8 @@ protected void masterOperation(
         final ClusterState state,
         final ActionListener<ClusterUpdateSettingsResponse> listener
     ) {
-        submitUnbatchedTask(UPDATE_TASK_SOURCE, new ClusterUpdateSettingsTask(clusterSettings, Priority.IMMEDIATE, request, listener) {
+        var storedContext = threadPool.getThreadContext().newRestorableContext(false, true);
+        var updateSettingsTask = new ClusterUpdateSettingsTask(clusterSettings, Priority.IMMEDIATE, request, storedContext, listener) {
             @Override
             protected ClusterUpdateSettingsResponse newResponse(boolean acknowledged) {
                 return new ClusterUpdateSettingsResponse(acknowledged, updater.getTransientUpdates(), updater.getPersistentUpdate());
@@ -228,35 +231,41 @@ public void onFailure(Exception e) {
                 logger.debug(() -> "failed to perform [" + UPDATE_TASK_SOURCE + "]", e);
                 super.onFailure(e);
             }
-        });
+        };
+        submitUnbatchedTask(UPDATE_TASK_SOURCE, updateSettingsTask);
     }
 
     private static class ClusterUpdateSettingsTask extends AckedClusterStateUpdateTask {
         protected volatile boolean reroute = false;
         protected final SettingsUpdater updater;
         protected final ClusterUpdateSettingsRequest request;
+        private final Supplier<ThreadContext.StoredContext> storedContext;
 
         ClusterUpdateSettingsTask(
             final ClusterSettings clusterSettings,
             Priority priority,
             ClusterUpdateSettingsRequest request,
+            Supplier<ThreadContext.StoredContext> storedContext,
             ActionListener<? extends AcknowledgedResponse> listener
         ) {
             super(priority, request, listener);
             this.updater = new SettingsUpdater(clusterSettings);
             this.request = request;
+            this.storedContext = storedContext;
         }
 
         @Override
         public ClusterState execute(final ClusterState currentState) {
-            final ClusterState clusterState = updater.updateSettings(
-                currentState,
-                request.transientSettings(),
-                request.persistentSettings(),
-                logger
-            );
-            reroute = clusterState != currentState;
-            return clusterState;
+            try (var ignored = storedContext.get()) {
+                final ClusterState clusterState = updater.updateSettings(
+                    currentState,
+                    request.transientSettings(),
+                    request.persistentSettings(),
+                    logger
+                );
+                reroute = clusterState != currentState;
+                return clusterState;
+            }
         }
     }
 

server/src/main/java/org/elasticsearch/common/util/concurrent/ThreadContext.java

@@ -449,27 +449,62 @@ private StoredContext newStoredContext(
      *     new Thread() {
      *         public void run() {
      *             try (ThreadContext.StoredContext ctx = restorable.get()) {
-     *                 // execute with the parents context and restore the threads context afterwards
+     *                 // execute with the parents context and restore the thread's context afterwards
+     *                 // response headers of this child context will be propagated back to the parent context
+     *                 // if `propagateResponseHeaders` is true
      *             }
      *         }
      *
      *     }.start();
      * </pre>
      *
      * @param preserveResponseHeaders if set to <code>true</code> the response headers of the restore thread will be preserved.
+     * @param propagateResponseHeaders if set to <code>true</code> the response headers of the child context will be propagated back.
      * @return a restorable context supplier
      */
-    public Supplier<StoredContext> newRestorableContext(boolean preserveResponseHeaders) {
-        return wrapRestorable(preserveResponseHeaders ? newStoredContextPreservingResponseHeaders() : newStoredContext());
+    public Supplier<StoredContext> newRestorableContext(boolean preserveResponseHeaders, boolean propagateResponseHeaders) {
+        return wrapRestorable(
+            preserveResponseHeaders ? newStoredContextPreservingResponseHeaders() : newStoredContext(),
+            propagateResponseHeaders
+        );
+    }
+
+    /**
+     * Returns a supplier that gathers a {@link #newStoredContextPreservingResponseHeaders()} and restores it once the
+     * returned supplier is invoked. The context returned from the supplier is a stored version of the
+     * suppliers callers context that should be restored once the originally gathered context is not needed anymore.
+     * For instance this method should be used like this:
+     *
+     * <pre>
+     *     Supplier&lt;ThreadContext.StoredContext&gt; restorable = context.newRestorableContext(true);
+     *     new Thread() {
+     *         public void run() {
+     *             try (ThreadContext.StoredContext ctx = restorable.get()) {
+     *                 // execute with the parents context and restore the thread's context afterwards
+     *             }
+     *         }
+     *
+     *     }.start();
+     * </pre>
+     *
+     * @param preserveExistingResponseHeaders if set to <code>true</code> the response headers of the restore thread will be preserved.
+     * @return a restorable context supplier
+     */
+    public Supplier<StoredContext> newRestorableContext(boolean preserveExistingResponseHeaders) {
+        return newRestorableContext(preserveExistingResponseHeaders, false);
     }
 
     /**
      * Same as {@link #newRestorableContext(boolean)} but wraps an existing context to restore.
      * @param storedContext the context to restore
      */
     public Supplier<StoredContext> wrapRestorable(StoredContext storedContext) {
+        return wrapRestorable(storedContext, false);
+    }
+
+    private Supplier<StoredContext> wrapRestorable(StoredContext storedContext, boolean propagateResponseHeaders) {
         return () -> {
-            StoredContext context = newStoredContext();
+            var context = propagateResponseHeaders ? newStoredContextPreservingResponseHeaders() : newStoredContext();
             storedContext.restore();
             return context;
         };