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
}
            };
            builder.putHandler(new ReservedStateHandlerMetadata(handler.name(), Set.of("a", "b")));
            handlers.add(handler);
            chunkMap.put(handler.name(), i);
        }
        final ReservedStateMetadata operatorMetadata = ReservedStateMetadata.builder("namespace_one").version(1L).build();
        Metadata metadata = Metadata.builder().put(operatorMetadata).build();
        ClusterState state = ClusterState.builder(new ClusterName("test")).metadata(metadata).build();
        var chunk = new ReservedStateChunk(chunkMap, new ReservedStateVersion(2L, Version.CURRENT));
        ClusterService clusterService = mock(ClusterService.class);
        final var controller = spy(new ReservedClusterStateService(clusterService, handlers));
        var trialRunResult = controller.trialRun(
            "namespace_one",
            state,
            chunk,
            new LinkedHashSet<>(handlers.stream().map(h -> h.name()).toList())
        );
        assertEquals(count, trialRunResult.nonStateTransforms().size());
        controller.executeNonStateTransformationSteps(trialRunResult.nonStateTransforms(), new ActionListener<>() {
            @Override
            public void onResponse(Collection<NonStateTransformResult> nonStateTransformResults) {
                assertEquals(count, nonStateTransformResults.size());
                var expectedHandlers = new ArrayList<String>();
                var expectedValues = new ArrayList<String>();
                for (int i = 0; i < count; i++) {
                    expectedHandlers.add("non-state:" + i);
                    expectedValues.add("key non-state:" + i);
                }
                assertThat(