public static HighAvailabilityServices createAvailableOrEmbeddedServices(
            Configuration config, Executor executor, FatalErrorHandler fatalErrorHandler)
            throws Exception {
        HighAvailabilityMode highAvailabilityMode = HighAvailabilityMode.fromConfig(config);

        switch (highAvailabilityMode) {
            case NONE:
                return new EmbeddedHaServices(executor);

            case ZOOKEEPER:
                return createZooKeeperHaServices(config, executor, fatalErrorHandler);

            case KUBERNETES:
                return createCustomHAServices(
                        "org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory",
                        config,
                        executor);

            case FACTORY_CLASS:
                return createCustomHAServices(config, executor);

            default:
                throw new Exception(
                        "High availability mode " + highAvailabilityMode + " is not supported.");
        }
    }
@Test(expected = Exception.class)
    public void testCustomHAServicesFactoryNotDefined() throws Exception {
        Configuration config = new Configuration();

        Executor executor = Executors.directExecutor();

        config.set(
                HighAvailabilityOptions.HA_MODE,
                HighAvailabilityMode.FACTORY_CLASS.name().toLowerCase());

        // expect
        HighAvailabilityServicesUtils.createAvailableOrEmbeddedServices(
                config, executor, NoOpFatalErrorHandler.INSTANCE);
    }