Merge pull request #15735 from jasontedor/master-node-change-predicate

Refactor master node change predicate for reuse

core/src/main/java/org/elasticsearch/action/support/master/TransportMasterNodeAction.java

@@ -26,10 +26,10 @@ import org.elasticsearch.action.ActionRunnable;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.action.support.ThreadedActionListener;
-import org.elasticsearch.cluster.ClusterChangedEvent;
 import org.elasticsearch.cluster.ClusterService;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.ClusterStateObserver;
+import org.elasticsearch.cluster.MasterNodeChangePredicate;
 import org.elasticsearch.cluster.NotMasterException;
 import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
@@ -51,20 +51,6 @@ import java.util.function.Supplier;
  * A base class for operations that needs to be performed on the master node.
  */
 public abstract class TransportMasterNodeAction<Request extends MasterNodeRequest, Response extends ActionResponse> extends HandledTransportAction<Request, Response> {
-    private static final ClusterStateObserver.ChangePredicate masterNodeChangedPredicate = new ClusterStateObserver.ChangePredicate() {
-        @Override
-        public boolean apply(ClusterState previousState, ClusterState.ClusterStateStatus previousStatus,
-                             ClusterState newState, ClusterState.ClusterStateStatus newStatus) {
-            // The condition !newState.nodes().masterNodeId().equals(previousState.nodes().masterNodeId()) is not sufficient as the same master node might get reelected after a disruption.
-            return newState.nodes().masterNodeId() != null && newState != previousState;
-        }
-
-        @Override
-        public boolean apply(ClusterChangedEvent event) {
-            return event.nodesDelta().masterNodeChanged();
-        }
-    };
-
     protected final TransportService transportService;
     protected final ClusterService clusterService;
 
@@ -164,7 +150,7 @@ public abstract class TransportMasterNodeAction<Request extends MasterNodeReques
                             if (t instanceof Discovery.FailedToCommitClusterStateException
                                     || (t instanceof NotMasterException)) {
                                 logger.debug("master could not publish cluster state or stepped down before publishing action [{}], scheduling a retry", t, actionName);
-                                retry(t, masterNodeChangedPredicate);
+                                retry(t, MasterNodeChangePredicate.INSTANCE);
                             } else {
                                 listener.onFailure(t);
                             }
@@ -180,7 +166,7 @@ public abstract class TransportMasterNodeAction<Request extends MasterNodeReques
             } else {
                 if (nodes.masterNode() == null) {
                     logger.debug("no known master node, scheduling a retry");
-                    retry(null, masterNodeChangedPredicate);
+                    retry(null, MasterNodeChangePredicate.INSTANCE);
                 } else {
                     transportService.sendRequest(nodes.masterNode(), actionName, request, new ActionListenerResponseHandler<Response>(listener) {
                         @Override
@@ -195,7 +181,7 @@ public abstract class TransportMasterNodeAction<Request extends MasterNodeReques
                                 // we want to retry here a bit to see if a new master is elected
                                 logger.debug("connection exception while trying to forward request with action name [{}] to master node [{}], scheduling a retry. Error: [{}]",
                                         actionName, nodes.masterNode(), exp.getDetailedMessage());
-                                retry(cause, masterNodeChangedPredicate);
+                                retry(cause, MasterNodeChangePredicate.INSTANCE);
                             } else {
                                 listener.onFailure(exp);
                             }

core/src/main/java/org/elasticsearch/cluster/MasterNodeChangePredicate.java

@@ -0,0 +1,40 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.cluster;
+
+public enum MasterNodeChangePredicate implements ClusterStateObserver.ChangePredicate {
+    INSTANCE;
+
+    @Override
+    public boolean apply(
+        ClusterState previousState,
+        ClusterState.ClusterStateStatus previousStatus,
+        ClusterState newState,
+        ClusterState.ClusterStateStatus newStatus) {
+        // checking if the masterNodeId changed is insufficient as the
+        // same master node might get re-elected after a disruption
+        return newState.nodes().masterNodeId() != null && newState != previousState;
+    }
+
+    @Override
+    public boolean apply(ClusterChangedEvent changedEvent) {
+        return changedEvent.nodesDelta().masterNodeChanged();
+    }
+}