Thread safe clean up of LocalNodeModeListeners (#60007)

This commit continues on the work in #59801 and makes other
implementors of the LocalNodeMasterListener interface thread safe in
that they will no longer allow the callbacks to run on different
threads and possibly race each other. This also helps address other
issues where these events could be queued to wait for execution while
the service keeps moving forward thinking it is the master even when
that is not the case.

In order to accomplish this, the LocalNodeMasterListener no longer has
the executorName() method to prevent future uses that could encounter
this surprising behavior.

Each use was inspected and if the class was also a
ClusterStateListener, the implementation of LocalNodeMasterListener
was removed in favor of a single listener that combined the logic. A
single listener is used and there is currently no guarantee on execution
order between ClusterStateListeners and LocalNodeMasterListeners,
so a future change there could cause undesired consequences. For other
classes, the implementations of the callbacks were inspected and if the
operations were lightweight, the overriden executorName method was
removed to use the default, which runs on the same thread.

Backport of #59932

Author: jaymode

server/src/main/java/org/elasticsearch/cluster/InternalClusterInfoService.java

@@ -111,9 +111,6 @@ public InternalClusterInfoService(Settings settings, ClusterService clusterServi
         clusterSettings.addSettingsUpdateConsumer(INTERNAL_CLUSTER_INFO_UPDATE_INTERVAL_SETTING, this::setUpdateFrequency);
         clusterSettings.addSettingsUpdateConsumer(DiskThresholdSettings.CLUSTER_ROUTING_ALLOCATION_DISK_THRESHOLD_ENABLED_SETTING,
                                                   this::setEnabled);
-
-        // listen for state changes (this node starts/stops being the elected master, or new nodes are added)
-        clusterService.addListener(this);
     }
 
     private void setEnabled(boolean enabled) {

server/src/main/java/org/elasticsearch/cluster/LocalNodeMasterListener.java

@@ -33,20 +33,5 @@ public interface LocalNodeMasterListener {
      * Called when the local node used to be the master, a new master was elected and it's no longer the local node.
      */
     void offMaster();
-
-    /**
-     * The name of the executor that the implementation of the callbacks of this lister should be executed on. The thread
-     * that is responsible for managing instances of this lister is the same thread handling the cluster state events. If
-     * the work done is the callbacks above is inexpensive, this value may be
-     * {@link org.elasticsearch.threadpool.ThreadPool.Names#SAME SAME} (indicating that the callbacks will run on the same thread
-     * as the cluster state events are fired with). On the other hand, if the logic in the callbacks are heavier and take
-     * longer to process (or perhaps involve blocking due to IO operations), prefer to execute them on a separate more appropriate
-     * executor (eg. {@link org.elasticsearch.threadpool.ThreadPool.Names#GENERIC GENERIC}
-     * or {@link org.elasticsearch.threadpool.ThreadPool.Names#MANAGEMENT MANAGEMENT}).
-     *
-     * @return The name of the executor that will run the callbacks of this listener.
-     */
-    String executorName();
-
 }
 

server/src/main/java/org/elasticsearch/cluster/service/ClusterApplierService.java

@@ -114,7 +114,7 @@ public ClusterApplierService(String nodeName, Settings settings, ClusterSettings
         this.clusterSettings = clusterSettings;
         this.threadPool = threadPool;
         this.state = new AtomicReference<>();
-        this.localNodeMasterListeners = new LocalNodeMasterListeners(threadPool);
+        this.localNodeMasterListeners = new LocalNodeMasterListeners();
         this.nodeName = nodeName;
 
         this.slowTaskLoggingThreshold = CLUSTER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING.get(settings);
@@ -611,29 +611,30 @@ public void run() {
     private static class LocalNodeMasterListeners implements ClusterStateListener {
 
         private final List<LocalNodeMasterListener> listeners = new CopyOnWriteArrayList<>();
-        private final ThreadPool threadPool;
         private volatile boolean master = false;
 
-        private LocalNodeMasterListeners(ThreadPool threadPool) {
-            this.threadPool = threadPool;
+        private LocalNodeMasterListeners() {
         }
 
         @Override
         public void clusterChanged(ClusterChangedEvent event) {
             if (!master && event.localNodeMaster()) {
                 master = true;
                 for (LocalNodeMasterListener listener : listeners) {
-                    java.util.concurrent.Executor executor = threadPool.executor(listener.executorName());
-                    executor.execute(new OnMasterRunnable(listener));
+                    try {
+                        listener.onMaster();
+                    } catch (Exception e) {
+                        logger.warn("failed to notify LocalNodeMasterListener", e);
+                    }
                 }
-                return;
-            }
-
-            if (master && !event.localNodeMaster()) {
+            } else if (master && !event.localNodeMaster()) {
                 master = false;
                 for (LocalNodeMasterListener listener : listeners) {
-                    java.util.concurrent.Executor executor = threadPool.executor(listener.executorName());
-                    executor.execute(new OffMasterRunnable(listener));
+                    try {
+                        listener.offMaster();
+                    } catch (Exception e) {
+                        logger.warn("failed to notify LocalNodeMasterListener", e);
+                    }
                 }
             }
         }
@@ -644,34 +645,6 @@ private void add(LocalNodeMasterListener listener) {
 
     }
 
-    private static class OnMasterRunnable implements Runnable {
-
-        private final LocalNodeMasterListener listener;
-
-        private OnMasterRunnable(LocalNodeMasterListener listener) {
-            this.listener = listener;
-        }
-
-        @Override
-        public void run() {
-            listener.onMaster();
-        }
-    }
-
-    private static class OffMasterRunnable implements Runnable {
-
-        private final LocalNodeMasterListener listener;
-
-        private OffMasterRunnable(LocalNodeMasterListener listener) {
-            this.listener = listener;
-        }
-
-        @Override
-        public void run() {
-            listener.offMaster();
-        }
-    }
-
     // this one is overridden in tests so we can control time
     protected long currentTimeInMillis() {
         return threadPool.relativeTimeInMillis();

server/src/main/java/org/elasticsearch/common/settings/ConsistentSettingsService.java

@@ -29,8 +29,10 @@
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.UUIDs;
 import org.elasticsearch.common.hash.MessageDigests;
-import org.elasticsearch.threadpool.ThreadPool;
 
+import javax.crypto.SecretKey;
+import javax.crypto.SecretKeyFactory;
+import javax.crypto.spec.PBEKeySpec;
 import java.nio.charset.StandardCharsets;
 import java.security.NoSuchAlgorithmException;
 import java.security.spec.InvalidKeySpecException;
@@ -45,10 +47,6 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Consumer;
 
-import javax.crypto.SecretKey;
-import javax.crypto.SecretKeyFactory;
-import javax.crypto.spec.PBEKeySpec;
-
 /**
  * Used to publish secure setting hashes in the cluster state and to validate those hashes against the local values of those same settings.
  * This is colloquially referred to as the secure setting consistency check. It will publish and verify hashes only for the collection
@@ -247,11 +245,6 @@ public void onFailure(String source, Exception e) {
         public void offMaster() {
             logger.trace("I am no longer master, nothing to do");
         }
-
-        @Override
-        public String executorName() {
-            return ThreadPool.Names.SAME;
-        }
     }
 
 }

server/src/main/java/org/elasticsearch/node/Node.java

@@ -1137,7 +1137,10 @@ private List<NetworkService.CustomNameResolver> getCustomNameResolvers(List<Disc
     /** Constructs a ClusterInfoService which may be mocked for tests. */
     protected ClusterInfoService newClusterInfoService(Settings settings, ClusterService clusterService,
                                                        ThreadPool threadPool, NodeClient client) {
-        return new InternalClusterInfoService(settings, clusterService, threadPool, client);
+        final InternalClusterInfoService service = new InternalClusterInfoService(settings, clusterService, threadPool, client);
+        // listen for state changes (this node starts/stops being the elected master, or new nodes are added)
+        clusterService.addListener(service);
+        return service;
     }
 
     /** Constructs a {@link org.elasticsearch.http.HttpServerTransport} which may be mocked for tests. */

server/src/test/java/org/elasticsearch/cluster/InternalClusterInfoServiceSchedulingTests.java

@@ -82,6 +82,7 @@ protected PrioritizedEsThreadPoolExecutor createThreadPoolExecutor() {
 
         final FakeClusterInfoServiceClient client = new FakeClusterInfoServiceClient(threadPool);
         final InternalClusterInfoService clusterInfoService = new InternalClusterInfoService(settings, clusterService, threadPool, client);
+        clusterService.addListener(clusterInfoService);
         clusterInfoService.addListener(ignored -> {});
 
         clusterService.setNodeConnectionsService(ClusterServiceUtils.createNoOpNodeConnectionsService());

server/src/test/java/org/elasticsearch/cluster/service/ClusterApplierServiceTests.java

@@ -293,11 +293,6 @@ public void onMaster() {
             public void offMaster() {
                 isMaster.set(false);
             }
-
-            @Override
-            public String executorName() {
-                return ThreadPool.Names.SAME;
-            }
         });
 
         ClusterState state = timedClusterApplierService.state();

test/framework/src/main/java/org/elasticsearch/node/MockNode.java

@@ -174,7 +174,9 @@ protected ClusterInfoService newClusterInfoService(Settings settings, ClusterSer
         if (getPluginsService().filterPlugins(MockInternalClusterInfoService.TestPlugin.class).isEmpty()) {
             return super.newClusterInfoService(settings, clusterService, threadPool, client);
         } else {
-            return new MockInternalClusterInfoService(settings, clusterService, threadPool, client);
+            final MockInternalClusterInfoService service = new MockInternalClusterInfoService(settings, clusterService, threadPool, client);
+            clusterService.addListener(service);
+            return service;
         }
     }
 

x-pack/plugin/enrich/src/main/java/org/elasticsearch/xpack/enrich/EnrichPolicyMaintenanceService.java

@@ -94,11 +94,6 @@ public void offMaster() {
         }
     }
 
-    @Override
-    public String executorName() {
-        return ThreadPool.Names.GENERIC;
-    }
-
     private void scheduleNext() {
         if (isMaster) {
             try {

x-pack/plugin/ilm/src/main/java/org/elasticsearch/xpack/ilm/IndexLifecycleService.java

@@ -15,7 +15,6 @@
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.ClusterStateApplier;
 import org.elasticsearch.cluster.ClusterStateListener;
-import org.elasticsearch.cluster.LocalNodeMasterListener;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Priority;
@@ -51,7 +50,7 @@
  * A service which runs the {@link LifecyclePolicy}s associated with indexes.
  */
 public class IndexLifecycleService
-    implements ClusterStateListener, ClusterStateApplier, SchedulerEngine.Listener, Closeable, LocalNodeMasterListener, IndexEventListener {
+    implements ClusterStateListener, ClusterStateApplier, SchedulerEngine.Listener, Closeable, IndexEventListener {
     private static final Logger logger = LogManager.getLogger(IndexLifecycleService.class);
     private static final Set<String> IGNORE_STEPS_MAINTENANCE_REQUESTED = Collections.singleton(ShrinkStep.NAME);
     private volatile boolean isMaster = false;
@@ -82,7 +81,6 @@ public IndexLifecycleService(Settings settings, Client client, ClusterService cl
         this.pollInterval = LifecycleSettings.LIFECYCLE_POLL_INTERVAL_SETTING.get(settings);
         clusterService.addStateApplier(this);
         clusterService.addListener(this);
-        clusterService.addLocalNodeMasterListener(this);
         clusterService.getClusterSettings().addSettingsUpdateConsumer(LifecycleSettings.LIFECYCLE_POLL_INTERVAL_SETTING,
             this::updatePollInterval);
     }
@@ -121,13 +119,11 @@ public ClusterState moveClusterStateToPreviouslyFailedStep(ClusterState currentS
         return newState;
     }
 
-    @Override
-    public void onMaster() {
-        this.isMaster = true;
+    // package private for testing
+    void onMaster(ClusterState clusterState) {
         maybeScheduleJob();
 
-        ClusterState clusterState = clusterService.state();
-        IndexLifecycleMetadata currentMetadata = clusterState.metadata().custom(IndexLifecycleMetadata.TYPE);
+        final IndexLifecycleMetadata currentMetadata = clusterState.metadata().custom(IndexLifecycleMetadata.TYPE);
         if (currentMetadata != null) {
             OperationMode currentMode = currentMetadata.getOperationMode();
             if (OperationMode.STOPPED.equals(currentMode)) {
@@ -184,17 +180,6 @@ public void onMaster() {
         }
     }
 
-    @Override
-    public void offMaster() {
-        this.isMaster = false;
-        cancelJob();
-    }
-
-    @Override
-    public String executorName() {
-        return ThreadPool.Names.MANAGEMENT;
-    }
-
     @Override
     public void beforeIndexAddedToCluster(Index index, Settings indexSettings) {
         if (shouldParseIndexName(indexSettings)) {
@@ -237,7 +222,20 @@ private synchronized void maybeScheduleJob() {
 
     @Override
     public void clusterChanged(ClusterChangedEvent event) {
-        IndexLifecycleMetadata lifecycleMetadata = event.state().metadata().custom(IndexLifecycleMetadata.TYPE);
+        // Instead of using a LocalNodeMasterListener to track master changes, this service will
+        // track them here to avoid conditions where master listener events run after other
+        // listeners that depend on what happened in the master listener
+        final boolean prevIsMaster = this.isMaster;
+        if (prevIsMaster != event.localNodeMaster()) {
+            this.isMaster = event.localNodeMaster();
+            if (this.isMaster) {
+                onMaster(event.state());
+            } else {
+                cancelJob();
+            }
+        }
+
+        final IndexLifecycleMetadata lifecycleMetadata = event.state().metadata().custom(IndexLifecycleMetadata.TYPE);
         if (this.isMaster && lifecycleMetadata != null) {
             triggerPolicies(event.state(), true);
         }