public static CuratorFrameworkWithUnhandledErrorListener startCuratorFramework(
            Configuration configuration, FatalErrorHandler fatalErrorHandler) {
        checkNotNull(configuration, "configuration");
        String zkQuorum = configuration.getValue(HighAvailabilityOptions.HA_ZOOKEEPER_QUORUM);

        if (zkQuorum == null || StringUtils.isBlank(zkQuorum)) {
            throw new RuntimeException(
                    "No valid ZooKeeper quorum has been specified. "
                            + "You can specify the quorum via the configuration key '"
                            + HighAvailabilityOptions.HA_ZOOKEEPER_QUORUM.key()
                            + "'.");
        }

        int sessionTimeout =
                Math.toIntExact(
                        configuration
                                .get(HighAvailabilityOptions.ZOOKEEPER_SESSION_TIMEOUT)
                                .toMillis());

        int connectionTimeout =
                Math.toIntExact(
                        configuration
                                .get(HighAvailabilityOptions.ZOOKEEPER_CONNECTION_TIMEOUT)
                                .toMillis());

        int retryWait =
                Math.toIntExact(
                        configuration.get(HighAvailabilityOptions.ZOOKEEPER_RETRY_WAIT).toMillis());

        int maxRetryAttempts =
                configuration.get(HighAvailabilityOptions.ZOOKEEPER_MAX_RETRY_ATTEMPTS);

        String root = configuration.getValue(HighAvailabilityOptions.HA_ZOOKEEPER_ROOT);

        String namespace = configuration.getValue(HighAvailabilityOptions.HA_CLUSTER_ID);

        boolean disableSaslClient = configuration.get(SecurityOptions.ZOOKEEPER_SASL_DISABLE);

        ACLProvider aclProvider;

        ZkClientACLMode aclMode = ZkClientACLMode.fromConfig(configuration);

        if (disableSaslClient && aclMode == ZkClientACLMode.CREATOR) {
            String errorMessage =
                    "Cannot set ACL role to "
                            + ZkClientACLMode.CREATOR
                            + "  since SASL authentication is "
                            + "disabled through the "
                            + SecurityOptions.ZOOKEEPER_SASL_DISABLE.key()
                            + " property";
            LOG.warn(errorMessage);
            throw new IllegalConfigurationException(errorMessage);
        }

        if (aclMode == ZkClientACLMode.CREATOR) {
            LOG.info("Enforcing creator for ZK connections");
            aclProvider = new SecureAclProvider();
        } else {
            LOG.info("Enforcing default ACL for ZK connections");
            aclProvider = new DefaultACLProvider();
        }

        String rootWithNamespace = generateZookeeperPath(root, namespace);

        LOG.info("Using '{}' as Zookeeper namespace.", rootWithNamespace);

        boolean ensembleTracking =
                configuration.get(HighAvailabilityOptions.ZOOKEEPER_ENSEMBLE_TRACKING);

        final CuratorFrameworkFactory.Builder curatorFrameworkBuilder =
                CuratorFrameworkFactory.builder()
                        .connectString(zkQuorum)
                        .sessionTimeoutMs(sessionTimeout)
                        .connectionTimeoutMs(connectionTimeout)
                        .retryPolicy(new ExponentialBackoffRetry(retryWait, maxRetryAttempts))
                        // Curator prepends a '/' manually and throws an Exception if the
                        // namespace starts with a '/'.
                        .namespace(trimStartingSlash(rootWithNamespace))
                        .ensembleTracker(ensembleTracking)
                        .aclProvider(aclProvider);

        if (configuration.contains(HighAvailabilityOptions.ZOOKEEPER_CLIENT_AUTHORIZATION)) {
            Map<String, String> authMap =
                    configuration.get(HighAvailabilityOptions.ZOOKEEPER_CLIENT_AUTHORIZATION);
            List<AuthInfo> authInfos =
                    authMap.entrySet().stream()
                            .map(
                                    entry ->
                                            new AuthInfo(
                                                    entry.getKey(),
                                                    entry.getValue()
                                                            .getBytes(
                                                                    ConfigConstants
                                                                            .DEFAULT_CHARSET)))
                            .collect(Collectors.toList());
            curatorFrameworkBuilder.authorization(authInfos);
        }

        if (configuration.contains(HighAvailabilityOptions.ZOOKEEPER_MAX_CLOSE_WAIT)) {
            long maxCloseWait =
                    configuration.get(HighAvailabilityOptions.ZOOKEEPER_MAX_CLOSE_WAIT).toMillis();
            if (maxCloseWait < 0 || maxCloseWait > Integer.MAX_VALUE) {
                throw new IllegalConfigurationException(
                        "The value (%d ms) is out-of-range for %s. The milliseconds timeout is expected to be between 0 and %d ms.",
                        maxCloseWait,
                        HighAvailabilityOptions.ZOOKEEPER_MAX_CLOSE_WAIT.key(),
                        Integer.MAX_VALUE);
            }
            curatorFrameworkBuilder.maxCloseWaitMs((int) maxCloseWait);
        }

        if (configuration.contains(
                HighAvailabilityOptions.ZOOKEEPER_SIMULATED_SESSION_EXP_PERCENT)) {
            curatorFrameworkBuilder.simulatedSessionExpirationPercent(
                    configuration.get(
                            HighAvailabilityOptions.ZOOKEEPER_SIMULATED_SESSION_EXP_PERCENT));
        }

        if (configuration.get(HighAvailabilityOptions.ZOOKEEPER_TOLERATE_SUSPENDED_CONNECTIONS)) {
            curatorFrameworkBuilder.connectionStateErrorPolicy(
                    new SessionConnectionStateErrorPolicy());
        }
        return startCuratorFramework(curatorFrameworkBuilder, fatalErrorHandler);
    }
@Test
    void testStartCuratorFrameworkFailed() throws Exception {
        TestingFatalErrorHandler handler = new TestingFatalErrorHandler();
        String errorMsg = "unexpected exception";
        final CuratorFrameworkFactory.Builder curatorFrameworkBuilder =
                CuratorFrameworkFactory.builder()
                        .connectString("localhost:2181")
                        .retryPolicy(new ExponentialBackoffRetry(1, 1))
                        .zookeeperFactory(
                                (s, i, watcher, b) -> {
                                    throw new RuntimeException(errorMsg);
                                })
                        .namespace("flink");
        ZooKeeperUtils.startCuratorFramework(curatorFrameworkBuilder, handler);
        assertThat(handler.getErrorFuture().get()).hasMessage(errorMsg);
    }