@Test public void testCreateCustomHAServices() throws Exception { Configuration config = new Configuration(); HighAvailabilityServices haServices = Mockito.mock(HighAvailabilityServices.class); TestHAFactory.haServices = haServices; Executor executor = Mockito.mock(Executor.class); config.setString(HighAvailabilityOptions.HA_MODE, TestHAFactory.class.getName()); HighAvailabilityServices actualHaServices = HighAvailabilityServicesUtils.createAvailableOrEmbeddedServices(config, executor); assertSame(haServices, actualHaServices); actualHaServices = HighAvailabilityServicesUtils.createHighAvailabilityServices(config, executor, HighAvailabilityServicesUtils.AddressResolution.NO_ADDRESS_RESOLUTION); assertSame(haServices, actualHaServices); }
private static HighAvailabilityServices createCustomHAServices(Configuration config, Executor executor) throws FlinkException { final ClassLoader classLoader = Thread.currentThread().getContextClassLoader(); final String haServicesClassName = config.getString(HighAvailabilityOptions.HA_MODE); final HighAvailabilityServicesFactory highAvailabilityServicesFactory = InstantiationUtil.instantiate( haServicesClassName, HighAvailabilityServicesFactory.class, classLoader); try { return highAvailabilityServicesFactory.createHAServices(config, executor); } catch (Exception e) { throw new FlinkException( String.format( "Could not create the ha services from the instantiated HighAvailabilityServicesFactory %s.", haServicesClassName), e); } }
HighAvailabilityServicesUtils { private static HighAvailabilityServices createCustomHAServices(Configuration config, Executor executor) throws FlinkException { final ClassLoader classLoader = Thread.currentThread().getContextClassLoader(); final String haServicesClassName = config.getString(HighAvailabilityOptions.HA_MODE); final HighAvailabilityServicesFactory highAvailabilityServicesFactory = InstantiationUtil.instantiate( haServicesClassName, HighAvailabilityServicesFactory.class, classLoader); try { return highAvailabilityServicesFactory.createHAServices(config, executor); } catch (Exception e) { throw new FlinkException( String.format( "Could not create the ha services from the instantiated HighAvailabilityServicesFactory %s.", haServicesClassName), e); } } }
HighAvailabilityServicesUtils { private static HighAvailabilityServices createCustomHAServices(Configuration config, Executor executor) throws FlinkException { final ClassLoader classLoader = Thread.currentThread().getContextClassLoader(); final String haServicesClassName = config.getString(HighAvailabilityOptions.HA_MODE); final HighAvailabilityServicesFactory highAvailabilityServicesFactory = InstantiationUtil.instantiate( haServicesClassName, HighAvailabilityServicesFactory.class, classLoader); try { return highAvailabilityServicesFactory.createHAServices(config, executor); } catch (Exception e) { throw new FlinkException( String.format( "Could not create the ha services from the instantiated HighAvailabilityServicesFactory %s.", haServicesClassName), e); } }  }
HighAvailabilityServicesUtils { private static HighAvailabilityServices createCustomHAServices(Configuration config, Executor executor) throws FlinkException { final ClassLoader classLoader = Thread.currentThread().getContextClassLoader(); final String haServicesClassName = config.getString(HighAvailabilityOptions.HA_MODE); final HighAvailabilityServicesFactory highAvailabilityServicesFactory = InstantiationUtil.instantiate( haServicesClassName, HighAvailabilityServicesFactory.class, classLoader); try { return highAvailabilityServicesFactory.createHAServices(config, executor); } catch (Exception e) { throw new FlinkException( String.format( "Could not create the ha services from the instantiated HighAvailabilityServicesFactory %s.", haServicesClassName), e); } }  static HighAvailabilityServices createAvailableOrEmbeddedServices(
		Configuration config,
		Executor executor); static HighAvailabilityServices createHighAvailabilityServices(
		Configuration configuration,
		Executor executor,
		AddressResolution addressResolution); static Tuple2<String, Integer> getJobManagerAddress(Configuration configuration); }
HighAvailabilityServicesUtils { private static HighAvailabilityServices createCustomHAServices(Configuration config, Executor executor) throws FlinkException { final ClassLoader classLoader = Thread.currentThread().getContextClassLoader(); final String haServicesClassName = config.getString(HighAvailabilityOptions.HA_MODE); final HighAvailabilityServicesFactory highAvailabilityServicesFactory = InstantiationUtil.instantiate( haServicesClassName, HighAvailabilityServicesFactory.class, classLoader); try { return highAvailabilityServicesFactory.createHAServices(config, executor); } catch (Exception e) { throw new FlinkException( String.format( "Could not create the ha services from the instantiated HighAvailabilityServicesFactory %s.", haServicesClassName), e); } }  static HighAvailabilityServices createAvailableOrEmbeddedServices(
		Configuration config,
		Executor executor); static HighAvailabilityServices createHighAvailabilityServices(
		Configuration configuration,
		Executor executor,
		AddressResolution addressResolution); static Tuple2<String, Integer> getJobManagerAddress(Configuration configuration);  }