public static HybridPartitionDataConsumeConstraint
            getOrDecideHybridPartitionDataConsumeConstraint(
                    Configuration configuration, boolean enableSpeculativeExecution) {
        final HybridPartitionDataConsumeConstraint hybridPartitionDataConsumeConstraint =
                configuration
                        .getOptional(JobManagerOptions.HYBRID_PARTITION_DATA_CONSUME_CONSTRAINT)
                        .orElseGet(
                                () -> {
                                    HybridPartitionDataConsumeConstraint defaultConstraint =
                                            enableSpeculativeExecution
                                                    ? ONLY_FINISHED_PRODUCERS
                                                    : UNFINISHED_PRODUCERS;
                                    LOG.info(
                                            "Set {} to {} as it is not configured",
                                            JobManagerOptions
                                                    .HYBRID_PARTITION_DATA_CONSUME_CONSTRAINT
                                                    .key(),
                                            defaultConstraint.name());
                                    return defaultConstraint;
                                });
        if (enableSpeculativeExecution) {
            Preconditions.checkState(
                    hybridPartitionDataConsumeConstraint != UNFINISHED_PRODUCERS,
                    "For speculative execution, only supports consume finished partition now.");
        }
        return hybridPartitionDataConsumeConstraint;
    }
@Test
    void testNotOnlyConsumeFinishedPartitionWithSpeculativeEnable() {
        Configuration configuration = new Configuration();
        configuration.set(
                JobManagerOptions.HYBRID_PARTITION_DATA_CONSUME_CONSTRAINT, UNFINISHED_PRODUCERS);
        assertThatThrownBy(
                        () -> getOrDecideHybridPartitionDataConsumeConstraint(configuration, true))
                .isInstanceOf(IllegalStateException.class);
    }