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
@Override
            public Map<String, Object> fromXContent(XContentParser parser) throws IOException {
                return parser.map();
            }
        };
        ReservedStateHandlerMetadata hmOne = new ReservedStateHandlerMetadata("non-state", Set.of("a", "b"));
        ReservedStateErrorMetadata emOne = new ReservedStateErrorMetadata(
            2L,
            ReservedStateErrorMetadata.ErrorKind.VALIDATION,
            List.of("Test error 1", "Test error 2")
        );
        final ReservedStateMetadata operatorMetadata = ReservedStateMetadata.builder("namespace_one")
            .errorMetadata(emOne)
            .version(1L)
            .putHandler(hmOne)
            .build();
        Metadata metadata = Metadata.builder().put(operatorMetadata).build();
        ClusterState state = ClusterState.builder(new ClusterName("test")).metadata(metadata).build();
        var chunk = new ReservedStateChunk(Map.of("non-state", "two", "maker", "three"), new ReservedStateVersion(2L, Version.CURRENT));
        var orderedHandlers = List.of(exceptionThrower.name(), newStateMaker.name());
        ClusterService clusterService = mock(ClusterService.class);
        final var controller = spy(new ReservedClusterStateService(clusterService, List.of(newStateMaker, exceptionThrower)));
        var trialRunResult = controller.trialRun("namespace_one", state, chunk, new LinkedHashSet<>(orderedHandlers));
        assertEquals(1, trialRunResult.nonStateTransforms().size());
        assertEquals(0, trialRunResult.errors().size());
        trialRunResult.nonStateTransforms().get(0).accept(new ActionListener<>() {
            @Override
            public void onResponse(NonStateTransformResult nonStateTransformResult) {
                assertThat(nonStateTransformResult.updatedKeys(), containsInAnyOrder("key non-state"));
                assertEquals("non-state", nonStateTransformResult.handlerName());