@Override
    public CompletableFuture<Collection<TaskManagerLocation>> getPreferredLocations(
            final ExecutionVertexID executionVertexId,
            final Set<ExecutionVertexID> producersToIgnore) {

        checkNotNull(executionVertexId);
        checkNotNull(producersToIgnore);

        final Collection<TaskManagerLocation> preferredLocationsBasedOnState =
                getPreferredLocationsBasedOnState(executionVertexId);
        if (!preferredLocationsBasedOnState.isEmpty()) {
            return CompletableFuture.completedFuture(preferredLocationsBasedOnState);
        }

        return getPreferredLocationsBasedOnInputs(executionVertexId, producersToIgnore);
    }
@Test
    void testStateLocationsWillBeReturnedIfExist() {
        final TaskManagerLocation stateLocation = new LocalTaskManagerLocation();

        final TestingInputsLocationsRetriever.Builder locationRetrieverBuilder =
                new TestingInputsLocationsRetriever.Builder();

        final ExecutionVertexID consumerId = new ExecutionVertexID(new JobVertexID(), 0);
        final ExecutionVertexID producerId = new ExecutionVertexID(new JobVertexID(), 0);
        locationRetrieverBuilder.connectConsumerToProducer(consumerId, producerId);

        final TestingInputsLocationsRetriever inputsLocationsRetriever =
                locationRetrieverBuilder.build();

        inputsLocationsRetriever.markScheduled(producerId);

        final PreferredLocationsRetriever locationsRetriever =
                new DefaultPreferredLocationsRetriever(
                        id -> Optional.of(stateLocation), inputsLocationsRetriever);

        final CompletableFuture<Collection<TaskManagerLocation>> preferredLocations =
                locationsRetriever.getPreferredLocations(consumerId, Collections.emptySet());

        assertThat(preferredLocations.getNow(null)).containsExactly(stateLocation);
    }