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
ReservedStateHandlerMetadata hmOne = new ReservedStateHandlerMetadata("one", 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();
        assertFalse(ReservedClusterStateService.isNewError(operatorMetadata, 2L));
        assertFalse(ReservedClusterStateService.isNewError(operatorMetadata, 1L));
        assertTrue(ReservedClusterStateService.isNewError(operatorMetadata, 3L));
        assertTrue(ReservedClusterStateService.isNewError(null, 1L));
        var chunk = new ReservedStateChunk(Map.of("one", "two", "maker", "three"), new ReservedStateVersion(2L, Version.CURRENT));
        var orderedHandlers = List.of(exceptionThrower.name(), newStateMaker.name());
        
        
        
        ReservedStateUpdateTask task = new ReservedStateUpdateTask(
            "namespace_one",
            chunk,
            List.of(),
            Map.of(exceptionThrower.name(), exceptionThrower, newStateMaker.name(), newStateMaker),
            orderedHandlers,
            (clusterState, errorState) -> { assertFalse(ReservedClusterStateService.isNewError(operatorMetadata, errorState.version())); },
            new ActionListener<>() {