From c47340f2f1a9eae52279e0c70885ebe2675fcd03 Mon Sep 17 00:00:00 2001
From: Jason Tedor <jason@tedor.me>
Date: Sun, 3 Jan 2016 12:39:25 -0500
Subject: [PATCH] Refactor master node change predicate for reuse

This commit migrates a ClusterStateObserver.ChangePredicate for
detecting a master node change into a separate class for reuse
elsewhere.
---
 .../master/TransportMasterNodeAction.java     | 22 ++--------
 .../cluster/MasterNodeChangePredicate.java    | 40 +++++++++++++++++++
 2 files changed, 44 insertions(+), 18 deletions(-)
 create mode 100644 core/src/main/java/org/elasticsearch/cluster/MasterNodeChangePredicate.java

diff --git a/core/src/main/java/org/elasticsearch/action/support/master/TransportMasterNodeAction.java b/core/src/main/java/org/elasticsearch/action/support/master/TransportMasterNodeAction.java
index a06b11d9d58..f3fe2d30bb6 100644
--- a/core/src/main/java/org/elasticsearch/action/support/master/TransportMasterNodeAction.java
+++ b/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;
@@ -50,20 +50,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;
 
@@ -148,7 +134,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);
                             }
@@ -164,7 +150,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
@@ -179,7 +165,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);
                             }
diff --git a/core/src/main/java/org/elasticsearch/cluster/MasterNodeChangePredicate.java b/core/src/main/java/org/elasticsearch/cluster/MasterNodeChangePredicate.java
new file mode 100644
index 00000000000..6d91ec7fd27
--- /dev/null
+++ b/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();
+    }
+}