Today the `ReservedClusterStateService` decides whether or not to submit an update-error task depending on the content of the last-applied state as exposed by `ClusterService#state()`. This may happen concurrently with another state being published, so does not give a fully consistent view of the latest state. Instead it should check the state within the cluster state task executor, triggering a no-op update as appropriate.
server/src/test/java/org/elasticsearch/reservedstate/service/ReservedClusterStateServiceTests.java
new ClusterStateTaskExecutor.BatchExecutionContext<>(state, List.of(taskContext), () -> null)
        );
        assertEquals(state, newState);
        assertTrue(successCalled.get());
        verify(task, times(1)).execute(any());
        taskExecutor.clusterStatePublished(state);
        verify(rerouteService, times(1)).reroute(anyString(), any(), any());
    }
    public void testErrorStateTask() throws Exception {
        ClusterState state = ClusterState.builder(new ClusterName("test")).build();
        final var listenerCompleted = new AtomicBoolean(false);
        ReservedStateErrorTask task = spy(
            new ReservedStateErrorTask(
                new ErrorState("test", 1L, List.of("some parse error", "some io error"), ReservedStateErrorMetadata.ErrorKind.PARSING),
                new ActionListener<>() {
                    @Override
                    public void onResponse(ActionResponse.Empty empty) {
                        listenerCompleted.set(true);
                    }
                    @Override
                    public void onFailure(Exception e) {}
                }
            )
        );
        ReservedStateErrorTaskExecutor.TaskContext<ReservedStateErrorTask> taskContext =
            new ReservedStateErrorTaskExecutor.TaskContext<>() {
                @Override
                public ReservedStateErrorTask getTask() {
                    return task;
                }