@Override
    public boolean isInputConsumable(
            SchedulingExecutionVertex executionVertex,
            Set<ExecutionVertexID> verticesToSchedule,
            Map<ConsumedPartitionGroup, Boolean> consumableStatusCache) {
        for (ConsumedPartitionGroup consumedPartitionGroup :
                executionVertex.getConsumedPartitionGroups()) {

            if (!consumableStatusCache.computeIfAbsent(
                    consumedPartitionGroup,
                    (group) -> isConsumedPartitionGroupConsumable(group, verticesToSchedule))) {
                return false;
            }
        }
        return true;
    }
@Test
    void testHybridAndBlockingInputButBlockingInputNotFinished() {
        final TestingSchedulingTopology topology = new TestingSchedulingTopology();

        final List<TestingSchedulingExecutionVertex> producers1 =
                topology.addExecutionVertices().withParallelism(1).finish();

        final List<TestingSchedulingExecutionVertex> producers2 =
                topology.addExecutionVertices().withParallelism(1).finish();

        final List<TestingSchedulingExecutionVertex> consumer =
                topology.addExecutionVertices().withParallelism(1).finish();

        topology.connectAllToAll(producers1, consumer)
                .withResultPartitionState(ResultPartitionState.CREATED)
                .withResultPartitionType(ResultPartitionType.BLOCKING)
                .finish();

        topology.connectAllToAll(producers2, consumer)
                .withResultPartitionState(ResultPartitionState.CREATED)
                .withResultPartitionType(ResultPartitionType.HYBRID_FULL)
                .finish();

        DefaultInputConsumableDecider inputConsumableDecider =
                createDefaultInputConsumableDecider(
                        Collections.singleton(producers2.get(0).getId()), topology);

        assertThat(
                        inputConsumableDecider.isInputConsumable(
                                consumer.get(0), Collections.emptySet(), new HashMap<>()))
                .isFalse();
    }