private static HighAvailabilityServices createCustomHAServices(
            Configuration config, Executor executor) throws FlinkException {
        return createCustomHAServices(
                config.get(HighAvailabilityOptions.HA_MODE), config, executor);
    }
@Test
    public void testCreateCustomHAServices() throws Exception {
        Configuration config = new Configuration();

        HighAvailabilityServices haServices = new TestingHighAvailabilityServices();
        TestHAFactory.haServices = haServices;

        Executor executor = Executors.directExecutor();

        config.set(HighAvailabilityOptions.HA_MODE, TestHAFactory.class.getName());

        // when
        HighAvailabilityServices actualHaServices =
                HighAvailabilityServicesUtils.createAvailableOrEmbeddedServices(
                        config, executor, NoOpFatalErrorHandler.INSTANCE);

        // then
        assertSame(haServices, actualHaServices);

        // when
        actualHaServices =
                HighAvailabilityServicesUtils.createHighAvailabilityServices(
                        config,
                        executor,
                        AddressResolution.NO_ADDRESS_RESOLUTION,
                        RpcSystem.load(),
                        NoOpFatalErrorHandler.INSTANCE);

        // then
        assertSame(haServices, actualHaServices);
    }