Changes FailedToCommitClusterStateException to NotMasterException (elastic#135548)

Changes a FailedToCommitClusterStateException incorrectly thrown prior
to cluster state update publication to a NotMasterException

Relates to ES-13061

Author: joshua-adams-1

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

@@ -34,6 +34,10 @@ public NotMasterException(StreamInput in) throws IOException {
         super(in);
     }
 
+    public NotMasterException(String msg, Object... args) {
+        super(msg, args);
+    }
+
     public NotMasterException(String msg, Throwable cause, Object... args) {
         super(msg, cause, args);
     }

server/src/main/java/org/elasticsearch/cluster/coordination/Coordinator.java

@@ -24,6 +24,7 @@
 import org.elasticsearch.cluster.ClusterStatePublicationEvent;
 import org.elasticsearch.cluster.ClusterStateUpdateTask;
 import org.elasticsearch.cluster.LocalMasterServiceTask;
+import org.elasticsearch.cluster.NotMasterException;
 import org.elasticsearch.cluster.block.ClusterBlocks;
 import org.elasticsearch.cluster.coordination.ClusterFormationFailureHelper.ClusterFormationState;
 import org.elasticsearch.cluster.coordination.CoordinationMetadata.VotingConfigExclusion;
@@ -1552,7 +1553,7 @@ public void publish(
                             clusterStatePublicationEvent.getNewState().term()
                         )
                     );
-                    throw new FailedToCommitClusterStateException(
+                    throw new NotMasterException(
                         "node is no longer master for term "
                             + clusterStatePublicationEvent.getNewState().term()
                             + " while handling publication"
@@ -1638,8 +1639,8 @@ assert getLocalNode().equals(clusterState.getNodes().get(getLocalNode().getId())
                     }
                 }
             }
-        } catch (FailedToCommitClusterStateException failedToCommitClusterStateException) {
-            publishListener.onFailure(failedToCommitClusterStateException);
+        } catch (FailedToCommitClusterStateException | NotMasterException e) {
+            publishListener.onFailure(e);
         } catch (Exception e) {
             assert false : e; // all exceptions should already be caught and wrapped in a FailedToCommitClusterStateException
             logger.error(() -> "[" + clusterStatePublicationEvent.getSummary() + "] publishing unexpectedly failed", e);

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

@@ -415,13 +415,30 @@ public void onResponse(Void unused) {
 
                 @Override
                 public void onFailure(Exception exception) {
-                    if (exception instanceof FailedToCommitClusterStateException failedToCommitClusterStateException) {
+                    if (exception instanceof FailedToCommitClusterStateException || exception instanceof NotMasterException) {
                         final long notificationStartTime = threadPool.rawRelativeTimeInMillis();
                         final long version = newClusterState.version();
-                        logger.warn(() -> format("failing [%s]: failed to commit cluster state version [%s]", summary, version), exception);
+
+                        if (exception instanceof FailedToCommitClusterStateException) {
+                            logger.warn(
+                                () -> format("Failing [%s]: failed to commit cluster state version [%s]", summary, version),
+                                exception
+                            );
+                        } else {
+                            logger.debug(
+                                () -> format(
+                                    "Failing [%s]: node is no longer the master. Failed to publish cluster state version [%s]",
+                                    summary,
+                                    version
+                                ),
+                                exception
+                            );
+                        }
+
                         for (final var executionResult : executionResults) {
-                            executionResult.onPublishFailure(failedToCommitClusterStateException);
+                            executionResult.onPublishFailure(exception);
                         }
+
                         final long notificationMillis = threadPool.rawRelativeTimeInMillis() - notificationStartTime;
                         clusterStateUpdateStatsTracker.onPublicationFailure(
                             threadPool.rawRelativeTimeInMillis(),
@@ -985,11 +1002,18 @@ void onClusterStateUnchanged(ClusterState clusterState) {
             }
         }
 
-        void onPublishFailure(FailedToCommitClusterStateException e) {
+        void onPublishFailure(Exception e) {
+            assert e instanceof FailedToCommitClusterStateException || e instanceof NotMasterException : e;
             if (publishedStateConsumer == null && onPublicationSuccess == null) {
                 assert failure != null;
                 var taskFailure = failure;
-                failure = new FailedToCommitClusterStateException(e.getMessage(), e);
+
+                if (e instanceof FailedToCommitClusterStateException) {
+                    failure = new FailedToCommitClusterStateException(e.getMessage(), e);
+                } else {
+                    failure = new NotMasterException(e.getMessage(), e);
+                }
+
                 failure.addSuppressed(taskFailure);
                 notifyFailure();
                 return;