public static StreamExchangeMode getBatchStreamExchangeMode(
            ReadableConfig config, StreamExchangeMode requiredExchangeMode) {
        if (requiredExchangeMode == StreamExchangeMode.BATCH) {
            return StreamExchangeMode.BATCH;
        }

        final GlobalStreamExchangeMode globalExchangeMode =
                getGlobalStreamExchangeMode(config).orElse(null);
        if (globalExchangeMode == GlobalStreamExchangeMode.ALL_EDGES_BLOCKING) {
            return StreamExchangeMode.BATCH;
        }

        final BatchShuffleMode shuffleMode = config.get(ExecutionOptions.BATCH_SHUFFLE_MODE);
        if (shuffleMode == BatchShuffleMode.ALL_EXCHANGES_BLOCKING) {
            return StreamExchangeMode.BATCH;
        } else if (shuffleMode == BatchShuffleMode.ALL_EXCHANGES_HYBRID_FULL) {
            return StreamExchangeMode.HYBRID_FULL;
        } else if (shuffleMode == BatchShuffleMode.ALL_EXCHANGES_HYBRID_SELECTIVE) {
            return StreamExchangeMode.HYBRID_SELECTIVE;
        }

        return StreamExchangeMode.UNDEFINED;
    }
@Test
    void testBatchStreamExchangeMode() {
        final Configuration configuration = new Configuration();

        assertThat(getBatchStreamExchangeMode(configuration, null))
                .isEqualTo(StreamExchangeMode.BATCH);

        configuration.set(
                ExecutionOptions.BATCH_SHUFFLE_MODE, BatchShuffleMode.ALL_EXCHANGES_BLOCKING);
        assertThat(getBatchStreamExchangeMode(configuration, null))
                .isEqualTo(StreamExchangeMode.BATCH);

        configuration.set(
                ExecutionOptions.BATCH_SHUFFLE_MODE, BatchShuffleMode.ALL_EXCHANGES_HYBRID_FULL);
        assertThat(getBatchStreamExchangeMode(configuration, null))
                .isEqualTo(StreamExchangeMode.HYBRID_FULL);

        configuration.set(
                ExecutionOptions.BATCH_SHUFFLE_MODE,
                BatchShuffleMode.ALL_EXCHANGES_HYBRID_SELECTIVE);
        assertThat(getBatchStreamExchangeMode(configuration, null))
                .isEqualTo(StreamExchangeMode.HYBRID_SELECTIVE);

        configuration.set(
                ExecutionOptions.BATCH_SHUFFLE_MODE, BatchShuffleMode.ALL_EXCHANGES_PIPELINED);
        assertThat(getBatchStreamExchangeMode(configuration, null))
                .isEqualTo(StreamExchangeMode.UNDEFINED);

        configuration.set(
                ExecutionOptions.BATCH_SHUFFLE_MODE, BatchShuffleMode.ALL_EXCHANGES_PIPELINED);
        assertThat(getBatchStreamExchangeMode(configuration, StreamExchangeMode.BATCH))
                .isEqualTo(StreamExchangeMode.BATCH);
    }