Replace pre publication failed to commit cluster state exceptions

server/src/main/java/org/elasticsearch/ElasticsearchException.java

@@ -89,6 +89,10 @@ public class ElasticsearchException extends RuntimeException implements ToXConte
 
     private static final TransportVersion UNKNOWN_VERSION_ADDED = TransportVersion.zero();
 
+    private static final TransportVersion FAILED_TO_PUBLISH_CLUSTER_STATE_EXCEPTION_TRANSPORT_VERSION = TransportVersion.fromName(
+        "failed_to_publish_cluster_state_exception"
+    );
+
     /**
      * Passed in the {@link Params} of {@link #generateThrowableXContent(XContentBuilder, Params, Throwable)}
      * to control if the {@code caused_by} element should render. Unlike most parameters to {@code toXContent} methods this parameter is
@@ -2030,6 +2034,12 @@ private enum ElasticsearchExceptionHandle {
             NoMatchingProjectException::new,
             185,
             NO_MATCHING_PROJECT_EXCEPTION_VERSION
+        ),
+        FAILED_TO_PUBLISH_CLUSTER_STATE_EXCEPTION(
+            org.elasticsearch.cluster.coordination.FailedToPublishClusterStateException.class,
+            org.elasticsearch.cluster.coordination.FailedToPublishClusterStateException::new,
+            186,
+            FAILED_TO_PUBLISH_CLUSTER_STATE_EXCEPTION_TRANSPORT_VERSION
         );
 
         final Class<? extends ElasticsearchException> exceptionClass;

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/action/shard/ShardStateAction.java

@@ -25,6 +25,7 @@
 import org.elasticsearch.cluster.ClusterStateTaskListener;
 import org.elasticsearch.cluster.NotMasterException;
 import org.elasticsearch.cluster.coordination.FailedToCommitClusterStateException;
+import org.elasticsearch.cluster.coordination.FailedToPublishClusterStateException;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.cluster.metadata.ProjectId;
@@ -164,6 +165,7 @@ private void sendShardAction(
     private static final Class<?>[] MASTER_CHANNEL_EXCEPTIONS = new Class<?>[] {
         NotMasterException.class,
         ConnectTransportException.class,
+        FailedToPublishClusterStateException.class,
         FailedToCommitClusterStateException.class };
 
     private static boolean isMasterChannelException(Throwable exp) {
@@ -936,7 +938,7 @@ public StartedShardEntry getStartedShardEntry() {
         public void onFailure(Exception e) {
             if (e instanceof NotMasterException) {
                 logger.debug(() -> format("%s no longer master while starting shard [%s]", entry.shardId, entry));
-            } else if (e instanceof FailedToCommitClusterStateException) {
+            } else if (e instanceof FailedToPublishClusterStateException || e instanceof FailedToCommitClusterStateException) {
                 logger.debug(() -> format("%s unexpected failure while starting shard [%s]", entry.shardId, entry), e);
             } else {
                 logger.error(() -> format("%s unexpected failure while starting shard [%s]", entry.shardId, entry), e);

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"
@@ -1567,7 +1568,7 @@ public void publish(
                             clusterStatePublicationEvent.getSummary()
                         )
                     );
-                    throw new FailedToCommitClusterStateException("publication " + currentPublication.get() + " already in progress");
+                    throw new FailedToPublishClusterStateException("publication " + currentPublication.get() + " already in progress");
                 }
 
                 assert assertPreviousStateConsistency(clusterStatePublicationEvent);
@@ -1586,7 +1587,7 @@ assert getLocalNode().equals(clusterState.getNodes().get(getLocalNode().getId())
                 } catch (Exception e) {
                     logger.debug(() -> "[" + clusterStatePublicationEvent.getSummary() + "] publishing failed during context creation", e);
                     becomeCandidate("publication context creation");
-                    throw new FailedToCommitClusterStateException("publishing failed during context creation", e);
+                    throw new FailedToPublishClusterStateException("publishing failed during context creation", e);
                 }
 
                 try (Releasable ignored = publicationContext::decRef) {
@@ -1607,7 +1608,7 @@ assert getLocalNode().equals(clusterState.getNodes().get(getLocalNode().getId())
                             e
                         );
                         becomeCandidate("publication creation");
-                        throw new FailedToCommitClusterStateException("publishing failed while starting", e);
+                        throw new FailedToPublishClusterStateException("publishing failed while starting", e);
                     }
 
                     try {
@@ -1638,12 +1639,12 @@ assert getLocalNode().equals(clusterState.getNodes().get(getLocalNode().getId())
                     }
                 }
             }
-        } catch (FailedToCommitClusterStateException failedToCommitClusterStateException) {
-            publishListener.onFailure(failedToCommitClusterStateException);
+        } catch (FailedToPublishClusterStateException | FailedToCommitClusterStateException | NotMasterException e) {
+            publishListener.onFailure(e);
         } catch (Exception e) {
-            assert false : e; // all exceptions should already be caught and wrapped in a FailedToCommitClusterStateException
+            assert false : e; // all exceptions should already be caught and wrapped in a FailedToPublishClusterStateException |
             logger.error(() -> "[" + clusterStatePublicationEvent.getSummary() + "] publishing unexpectedly failed", e);
-            publishListener.onFailure(new FailedToCommitClusterStateException("publishing unexpectedly failed", e));
+            publishListener.onFailure(new FailedToPublishClusterStateException("publishing unexpectedly failed", e));
         }
     }
 

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

@@ -16,14 +16,19 @@
 import java.io.IOException;
 
 /**
- * Thrown when a cluster state publication fails to commit the new cluster state. If publication fails then a new master is elected but the
- * update might or might not take effect, depending on whether the newly-elected master accepted the published state that failed to
- * be committed. This exception should only be used when there is <i>ambiguity</i> whether a state update took effect or not.
- *
- * This is different from {@link NotMasterException} where we know for certain that a state update never took effect.
- *
- * This exception is retryable within {@link TransportMasterNodeAction}.
- *
+ * Exception indicating a cluster state update was published but not committed to all nodes.
+ * <p>
+ * If this exception is thrown, then the cluster state update was published, but is not guaranteed
+ * to be committed on any nodes, including the next master node. This exception should only be thrown when there is
+ * <i>ambiguity</i> whether a cluster state update has been committed.
+ * <p>
+ * For exceptions thrown prior to publication,
+ * when the cluster update has <i>definitely</i> failed, use a {@link FailedToPublishClusterStateException}.
+ * <p>
+ * If during a cluster state update the node is no longer master, use a {@link NotMasterException}
+ * <p>
+ * This is a retryable exception inside {@link TransportMasterNodeAction}
+ * <p>
  * See {@link ClusterStatePublisher} for more details.
  */
 public class FailedToCommitClusterStateException extends ElasticsearchException {

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

@@ -0,0 +1,46 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+package org.elasticsearch.cluster.coordination;
+
+import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.action.support.master.TransportMasterNodeAction;
+import org.elasticsearch.common.io.stream.StreamInput;
+
+import java.io.IOException;
+
+/**
+ * Exception indicating a cluster state update failed prior to publication.
+ * <p>
+ * If this exception is thrown, then the cluster state update was <i>not</i> published to any node.
+ * It is therefore impossible for the new master to have committed this state.
+ * <p>
+ * For exceptions thrown <i>after</i> publication, when the cluster state update may or may not have been committed,
+ * use a {@link FailedToCommitClusterStateException}.
+ * <p>
+ * This is a retryable exception inside {@link TransportMasterNodeAction}
+ */
+public class FailedToPublishClusterStateException extends ElasticsearchException {
+
+    public FailedToPublishClusterStateException(String msg) {
+        super(msg);
+    }
+
+    public FailedToPublishClusterStateException(StreamInput in) throws IOException {
+        super(in);
+    }
+
+    public FailedToPublishClusterStateException(String msg, Throwable cause, Object... args) {
+        super(msg, cause, args);
+    }
+
+    @Override
+    public Throwable fillInStackTrace() {
+        return this;
+    }
+}

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

@@ -26,6 +26,7 @@
 import org.elasticsearch.cluster.NotMasterException;
 import org.elasticsearch.cluster.coordination.ClusterStatePublisher;
 import org.elasticsearch.cluster.coordination.FailedToCommitClusterStateException;
+import org.elasticsearch.cluster.coordination.FailedToPublishClusterStateException;
 import org.elasticsearch.cluster.metadata.ProcessClusterEventTimeoutException;
 import org.elasticsearch.cluster.metadata.ProjectMetadata;
 import org.elasticsearch.cluster.node.DiscoveryNode;
@@ -415,13 +416,37 @@ public void onResponse(Void unused) {
 
                 @Override
                 public void onFailure(Exception exception) {
-                    if (exception instanceof FailedToCommitClusterStateException failedToCommitClusterStateException) {
+                    if (exception instanceof FailedToPublishClusterStateException
+                        || 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 if (exception instanceof FailedToPublishClusterStateException) {
+                            logger.warn(
+                                () -> format("failing [%s]: failed to publish cluster state version [%s]", summary, version),
+                                exception
+                            );
+                        } else {
+                            logger.debug(
+                                () -> format(
+                                    "node is no longer the master prior to publication of cluster state version [%s]: [%s]",
+                                    version,
+                                    summary
+                                ),
+                                exception
+                            );
+                        }
+
                         for (final var executionResult : executionResults) {
-                            executionResult.onPublishFailure(failedToCommitClusterStateException);
+                            executionResult.onPublishFailure(exception);
                         }
+
                         final long notificationMillis = threadPool.rawRelativeTimeInMillis() - notificationStartTime;
                         clusterStateUpdateStatsTracker.onPublicationFailure(
                             threadPool.rawRelativeTimeInMillis(),
@@ -985,11 +1010,19 @@ void onClusterStateUnchanged(ClusterState clusterState) {
             }
         }
 
-        void onPublishFailure(FailedToCommitClusterStateException e) {
+        void onPublishFailure(Exception e) {
             if (publishedStateConsumer == null && onPublicationSuccess == null) {
                 assert failure != null;
                 var taskFailure = failure;
-                failure = new FailedToCommitClusterStateException(e.getMessage(), e);
+
+                if (e instanceof FailedToPublishClusterStateException) {
+                    failure = new FailedToPublishClusterStateException(e.getMessage(), e);
+                } else if (e instanceof FailedToCommitClusterStateException) {
+                    failure = new FailedToCommitClusterStateException(e.getMessage(), e);
+                } else {
+                    failure = new NotMasterException(e.getMessage(), e);
+                }
+
                 failure.addSuppressed(taskFailure);
                 notifyFailure();
                 return;
@@ -1250,7 +1283,9 @@ synchronized ClusterStateUpdateStats getStatistics() {
     }
 
     public static boolean isPublishFailureException(Exception e) {
-        return e instanceof NotMasterException || e instanceof FailedToCommitClusterStateException;
+        return e instanceof NotMasterException
+            || e instanceof FailedToCommitClusterStateException
+            || e instanceof FailedToPublishClusterStateException;
     }
 
     private final Runnable queuesProcessor = new AbstractRunnable() {

server/src/main/java/org/elasticsearch/reservedstate/service/FileSettingsService.java

@@ -20,6 +20,7 @@
 import org.elasticsearch.cluster.ClusterStateListener;
 import org.elasticsearch.cluster.NotMasterException;
 import org.elasticsearch.cluster.coordination.FailedToCommitClusterStateException;
+import org.elasticsearch.cluster.coordination.FailedToPublishClusterStateException;
 import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.cluster.metadata.ReservedStateMetadata;
@@ -254,6 +255,8 @@ protected void onProcessFileChangesException(Path file, Exception e) {
             } else if (cause instanceof NotMasterException) {
                 logger().error(Strings.format("Node is no longer master while processing file [%s]", file), e);
                 return;
+            } else if (cause instanceof FailedToPublishClusterStateException) {
+                logger().error(Strings.format("Unable to publish cluster state while processing file [%s]", file), e);
             }
         }
 

server/src/main/java/org/elasticsearch/snapshots/SnapshotsService.java

@@ -35,6 +35,7 @@
 import org.elasticsearch.cluster.SnapshotsInProgress.ShardSnapshotStatus;
 import org.elasticsearch.cluster.SnapshotsInProgress.ShardState;
 import org.elasticsearch.cluster.coordination.FailedToCommitClusterStateException;
+import org.elasticsearch.cluster.coordination.FailedToPublishClusterStateException;
 import org.elasticsearch.cluster.metadata.DataStream;
 import org.elasticsearch.cluster.metadata.DataStreamAlias;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
@@ -1310,7 +1311,12 @@ private void handleFinalizationFailure(
         RepositoryData repositoryData,
         UpdatedShardGenerations updatedShardGenerations
     ) {
-        if (ExceptionsHelper.unwrap(e, NotMasterException.class, FailedToCommitClusterStateException.class) != null) {
+        if (ExceptionsHelper.unwrap(
+            e,
+            NotMasterException.class,
+            FailedToPublishClusterStateException.class,
+            FailedToCommitClusterStateException.class
+        ) != null) {
             // Failure due to not being master any more, don't try to remove snapshot from cluster state the next master
             // will try ending this snapshot again
             logger.debug(() -> "[" + snapshot + "] failed to update cluster state during snapshot finalization", e);
@@ -1962,7 +1968,12 @@ private void failAllListenersOnMasterFailOver(Exception e) {
         logger.debug("Failing all snapshot operation listeners because this node is not master any longer", e);
         final List<Runnable> readyToResolveListeners = new ArrayList<>();
         synchronized (currentlyFinalizing) {
-            if (ExceptionsHelper.unwrap(e, NotMasterException.class, FailedToCommitClusterStateException.class) != null) {
+            if (ExceptionsHelper.unwrap(
+                e,
+                NotMasterException.class,
+                FailedToPublishClusterStateException.class,
+                FailedToCommitClusterStateException.class
+            ) != null) {
                 repositoryOperations.clear();
                 for (final Snapshot snapshot : snapshotCompletionListeners.keySet()) {
                     failSnapshotCompletionListeners(

server/src/main/resources/transport/definitions/referable/failed_to_publish_cluster_state_exception.csv

@@ -0,0 +1 @@
+9187000

server/src/main/resources/transport/upper_bounds/9.3.csv

@@ -1 +1 @@
-esql_plan_with_no_columns,9186000
+failed_to_publish_cluster_state_exception,9187000

server/src/test/java/org/elasticsearch/ExceptionSerializationTests.java

@@ -25,6 +25,7 @@
 import org.elasticsearch.cluster.action.shard.ShardStateAction;
 import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.cluster.coordination.CoordinationStateRejectedException;
+import org.elasticsearch.cluster.coordination.FailedToPublishClusterStateException;
 import org.elasticsearch.cluster.coordination.NoMasterBlockService;
 import org.elasticsearch.cluster.coordination.NodeHealthCheckFailureException;
 import org.elasticsearch.cluster.desirednodes.VersionConflictException;
@@ -848,6 +849,7 @@ public void testIds() {
         ids.put(183, IndexDocFailureStoreStatus.ExceptionWithFailureStoreStatus.class);
         ids.put(184, RemoteException.class);
         ids.put(185, NoMatchingProjectException.class);
+        ids.put(186, FailedToPublishClusterStateException.class);
 
         Map<Class<? extends ElasticsearchException>, Integer> reverse = new HashMap<>();
         for (Map.Entry<Integer, Class<? extends ElasticsearchException>> entry : ids.entrySet()) {