@Deprecated
    static Optional<GlobalStreamExchangeMode> getGlobalStreamExchangeMode(ReadableConfig config) {
        return config.getOptional(ExecutionConfigOptions.TABLE_EXEC_SHUFFLE_MODE)
                .map(
                        value -> {
                            try {
                                return GlobalStreamExchangeMode.valueOf(
                                        convertLegacyShuffleMode(value).toUpperCase());
                            } catch (IllegalArgumentException e) {
                                throw new IllegalArgumentException(
                                        String.format(
                                                "Unsupported value %s for config %s.",
                                                value,
                                                ExecutionConfigOptions.TABLE_EXEC_SHUFFLE_MODE
                                                        .key()));
                            }
                        });
    }
@Test
    void testLegacyShuffleMode() {
        final Configuration configuration = new Configuration();

        configuration.set(
                ExecutionConfigOptions.TABLE_EXEC_SHUFFLE_MODE,
                GlobalStreamExchangeMode.ALL_EDGES_BLOCKING.toString());
        assertThat(getGlobalStreamExchangeMode(configuration).orElseThrow(AssertionError::new))
                .isEqualTo(GlobalStreamExchangeMode.ALL_EDGES_BLOCKING);

        configuration.set(
                ExecutionConfigOptions.TABLE_EXEC_SHUFFLE_MODE,
                GlobalStreamExchangeMode.FORWARD_EDGES_PIPELINED.toString());
        assertThat(getGlobalStreamExchangeMode(configuration).orElseThrow(AssertionError::new))
                .isEqualTo(GlobalStreamExchangeMode.FORWARD_EDGES_PIPELINED);

        configuration.set(
                ExecutionConfigOptions.TABLE_EXEC_SHUFFLE_MODE,
                GlobalStreamExchangeMode.POINTWISE_EDGES_PIPELINED.toString());
        assertThat(getGlobalStreamExchangeMode(configuration).orElseThrow(AssertionError::new))
                .isEqualTo(GlobalStreamExchangeMode.POINTWISE_EDGES_PIPELINED);

        configuration.set(
                ExecutionConfigOptions.TABLE_EXEC_SHUFFLE_MODE,
                GlobalStreamExchangeMode.ALL_EDGES_PIPELINED.toString());
        assertThat(getGlobalStreamExchangeMode(configuration).orElseThrow(AssertionError::new))
                .isEqualTo(GlobalStreamExchangeMode.ALL_EDGES_PIPELINED);

        configuration.set(
                ExecutionConfigOptions.TABLE_EXEC_SHUFFLE_MODE,
                StreamExchangeModeUtils.ALL_EDGES_BLOCKING_LEGACY);
        assertThat(getGlobalStreamExchangeMode(configuration).orElseThrow(AssertionError::new))
                .isEqualTo(GlobalStreamExchangeMode.ALL_EDGES_BLOCKING);

        configuration.set(
                ExecutionConfigOptions.TABLE_EXEC_SHUFFLE_MODE,
                StreamExchangeModeUtils.ALL_EDGES_PIPELINED_LEGACY);
        assertThat(getGlobalStreamExchangeMode(configuration).orElseThrow(AssertionError::new))
                .isEqualTo(GlobalStreamExchangeMode.ALL_EDGES_PIPELINED);

        configuration.set(
                ExecutionConfigOptions.TABLE_EXEC_SHUFFLE_MODE, "Forward_edges_PIPELINED");
        assertThat(
                        StreamExchangeModeUtils.getGlobalStreamExchangeMode(configuration)
                                .orElseThrow(AssertionError::new))
                .isEqualTo(GlobalStreamExchangeMode.FORWARD_EDGES_PIPELINED);
    }