public static String getZooKeeperEnsemble(Configuration flinkConf)
            throws IllegalConfigurationException {

        String zkQuorum = flinkConf.getValue(HighAvailabilityOptions.HA_ZOOKEEPER_QUORUM);

        if (zkQuorum == null || StringUtils.isBlank(zkQuorum)) {
            throw new IllegalConfigurationException("No ZooKeeper quorum specified in config.");
        }

        // Remove all whitespace
        zkQuorum = zkQuorum.replaceAll("\\s+", "");

        return zkQuorum;
    }
@Test
    void testZooKeeperEnsembleConnectStringConfiguration() throws Exception {
        // ZooKeeper does not like whitespace in the quorum connect String.
        String actual, expected;
        Configuration conf = new Configuration();

        {
            expected = "localhost:2891";

            setQuorum(conf, expected);
            actual = ZooKeeperUtils.getZooKeeperEnsemble(conf);
            assertThat(actual).isEqualTo(expected);

            setQuorum(conf, " localhost:2891 "); // with leading and trailing whitespace
            actual = ZooKeeperUtils.getZooKeeperEnsemble(conf);
            assertThat(actual).isEqualTo(expected);

            setQuorum(conf, "localhost :2891"); // whitespace after port
            actual = ZooKeeperUtils.getZooKeeperEnsemble(conf);
            assertThat(actual).isEqualTo(expected);
        }

        {
            expected = "localhost:2891,localhost:2891";

            setQuorum(conf, "localhost:2891,localhost:2891");
            actual = ZooKeeperUtils.getZooKeeperEnsemble(conf);
            assertThat(actual).isEqualTo(expected);

            setQuorum(conf, "localhost:2891, localhost:2891");
            actual = ZooKeeperUtils.getZooKeeperEnsemble(conf);
            assertThat(actual).isEqualTo(expected);

            setQuorum(conf, "localhost :2891, localhost:2891");
            actual = ZooKeeperUtils.getZooKeeperEnsemble(conf);
            assertThat(actual).isEqualTo(expected);

            setQuorum(conf, " localhost:2891, localhost:2891 ");
            actual = ZooKeeperUtils.getZooKeeperEnsemble(conf);
            assertThat(actual).isEqualTo(expected);
        }
    }