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 void onFailure(Exception e) {
                fail("Should not reach here");
            }
        });
    }
    public void testExecuteNonStateTransformationSteps() {
        int count = randomInt(10);
        var handlers = new ArrayList<ReservedClusterStateHandler<?>>();
        var i = 0;
        var builder = ReservedStateMetadata.builder("namespace_one").version(1L);
        var chunkMap = new HashMap<String, Object>();
        while (i < count) {
            final var key = i++;
            var handler = new ReservedClusterStateHandler<>() {
                @Override
                public String name() {
                    return "non-state:" + key;
                }
                @Override
                public TransformState transform(Object source, TransformState prevState) {
                    return new TransformState(prevState.state(), prevState.keys(), (l) -> internalKeys(l));
                }
                private void internalKeys(ActionListener<NonStateTransformResult> listener) {
                    listener.onResponse(new NonStateTransformResult(name(), Set.of("key non-state:" + key)));
                }
                @Override
                public Map<String, Object> fromXContent(XContentParser parser) throws IOException {
                    return parser.map();