flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java
/**
	 * Starts the mini cluster, based on the configured properties.
	 * 
	 * @throws Exception This method passes on any exception that occurs during the startup of
	 *                   the mini cluster.
	 */
Starts the mini cluster, based on the configured properties.

@throws Exception This method passes on any exception that occurs during the startup of
the mini cluster.
Starts the mini cluster based on the configured properties . @throws Exception This method passes on any exception that occurs during the startup of the mini cluster .
public void start() throws Exception {
		synchronized (lock) {
			checkState(!running, "FlinkMiniCluster is already running");
			LOG.info("Starting Flink Mini Cluster");
			LOG.debug("Using configuration {}", miniClusterConfiguration);
			final Configuration configuration = miniClusterConfiguration.getConfiguration();
			final Time rpcTimeout = miniClusterConfiguration.getRpcTimeout();
			final int numJobManagers = miniClusterConfiguration.getNumJobManagers();
			final int numTaskManagers = miniClusterConfiguration.getNumTaskManagers();
			final int numResourceManagers = miniClusterConfiguration.getNumResourceManagers();
			final boolean useSingleRpcService = miniClusterConfiguration.getRpcServiceSharing() == MiniClusterConfiguration.RpcServiceSharing.SHARED;
			try {
				LOG.info("Starting Metrics Registry");
				metricRegistry = createMetricRegistry(configuration);
				RpcService[] jobManagerRpcServices = new RpcService[numJobManagers];
				RpcService[] taskManagerRpcServices = new RpcService[numTaskManagers];
				RpcService[] resourceManagerRpcServices = new RpcService[numResourceManagers];
				LOG.info("Starting RPC Service(s)");
				commonRpcService = createRpcService(configuration, rpcTimeout, false, null);
				final ActorSystem actorSystem = ((AkkaRpcService) commonRpcService).getActorSystem();
				metricRegistry.startQueryService(actorSystem, null);
				if (useSingleRpcService) {
					for (int i = 0; i < numJobManagers; i++) {
						jobManagerRpcServices[i] = commonRpcService;
					}
					for (int i = 0; i < numTaskManagers; i++) {
						taskManagerRpcServices[i] = commonRpcService;
					}
					for (int i = 0; i < numResourceManagers; i++) {
						resourceManagerRpcServices[i] = commonRpcService;
					}
					this.resourceManagerRpcServices = null;
					this.jobManagerRpcServices = null;
					this.taskManagerRpcServices = null;
				}
				else {
					final String jobManagerBindAddress = miniClusterConfiguration.getJobManagerBindAddress();
					final String taskManagerBindAddress = miniClusterConfiguration.getTaskManagerBindAddress();
					final String resourceManagerBindAddress = miniClusterConfiguration.getResourceManagerBindAddress();
					for (int i = 0; i < numJobManagers; i++) {
						jobManagerRpcServices[i] = createRpcService(
								configuration, rpcTimeout, true, jobManagerBindAddress);
					}
					for (int i = 0; i < numTaskManagers; i++) {
						taskManagerRpcServices[i] = createRpcService(
								configuration, rpcTimeout, true, taskManagerBindAddress);
					}
					for (int i = 0; i < numResourceManagers; i++) {
						resourceManagerRpcServices[i] = createRpcService(
								configuration, rpcTimeout, true, resourceManagerBindAddress);
					}
					this.jobManagerRpcServices = jobManagerRpcServices;
					this.taskManagerRpcServices = taskManagerRpcServices;
					this.resourceManagerRpcServices = resourceManagerRpcServices;
				}
				LOG.info("Starting high-availability services");
				haServices = HighAvailabilityServicesUtils.createAvailableOrEmbeddedServices(
					configuration,
					commonRpcService.getExecutor());
				blobServer = new BlobServer(configuration, haServices.createBlobStore());
				blobServer.start();
				heartbeatServices = HeartbeatServices.fromConfiguration(configuration);
				LOG.info("Starting {} ResourceManger(s)", numResourceManagers);
				resourceManagerRunners = startResourceManagers(
					configuration,
					haServices,
					heartbeatServices,
					metricRegistry,
					numResourceManagers,
					resourceManagerRpcServices);
				LOG.info("Starting {} TaskManger(s)", numTaskManagers);
				taskManagers = startTaskManagers(
						configuration, haServices, metricRegistry, numTaskManagers, taskManagerRpcServices);
				LOG.info("Starting job dispatcher(s) for {} JobManger(s)", numJobManagers);
				jobDispatcher = new MiniClusterJobDispatcher(
					configuration,
					haServices,
					blobServer,
					heartbeatServices,
					metricRegistry,
					numJobManagers,
					jobManagerRpcServices);
			}
			catch (Exception e) {
				try {
					shutdownInternally();
				} catch (Exception ee) {
					e.addSuppressed(ee);
				}
				throw e;
			}
			running = true;
			LOG.info("Flink Mini Cluster started successfully");
		}
	}
public void start ( ) throws Exception { synchronized ( lock ) { checkState ( ! running , "FlinkMiniCluster is already running" ) ; LOG . info ( "Starting Flink Mini Cluster" ) ; LOG . debug ( "Using configuration {}" , miniClusterConfiguration ) ; final Configuration configuration = miniClusterConfiguration . getConfiguration ( ) ; final Time rpcTimeout = miniClusterConfiguration . getRpcTimeout ( ) ; final int numJobManagers = miniClusterConfiguration . getNumJobManagers ( ) ; final int numTaskManagers = miniClusterConfiguration . getNumTaskManagers ( ) ; final int numResourceManagers = miniClusterConfiguration . getNumResourceManagers ( ) ; final boolean useSingleRpcService = miniClusterConfiguration . getRpcServiceSharing ( ) == MiniClusterConfiguration . RpcServiceSharing . SHARED ; try { LOG . info ( "Starting Metrics Registry" ) ; metricRegistry = createMetricRegistry ( configuration ) ; RpcService [ ] jobManagerRpcServices = new RpcService [ numJobManagers ] ; RpcService [ ] taskManagerRpcServices = new RpcService [ numTaskManagers ] ; RpcService [ ] resourceManagerRpcServices = new RpcService [ numResourceManagers ] ; LOG . info ( "Starting RPC Service(s)" ) ; commonRpcService = createRpcService ( configuration , rpcTimeout , false , null ) ; final ActorSystem actorSystem = ( ( AkkaRpcService ) commonRpcService ) . getActorSystem ( ) ; metricRegistry . startQueryService ( actorSystem , null ) ; if ( useSingleRpcService ) { for ( int i = 0 ; i < numJobManagers ; i ++ ) { jobManagerRpcServices [ i ] = commonRpcService ; } for ( int i = 0 ; i < numTaskManagers ; i ++ ) { taskManagerRpcServices [ i ] = commonRpcService ; } for ( int i = 0 ; i < numResourceManagers ; i ++ ) { resourceManagerRpcServices [ i ] = commonRpcService ; } this . resourceManagerRpcServices = null ; this . jobManagerRpcServices = null ; this . taskManagerRpcServices = null ; } else { final String jobManagerBindAddress = miniClusterConfiguration . getJobManagerBindAddress ( ) ; final String taskManagerBindAddress = miniClusterConfiguration . getTaskManagerBindAddress ( ) ; final String resourceManagerBindAddress = miniClusterConfiguration . getResourceManagerBindAddress ( ) ; for ( int i = 0 ; i < numJobManagers ; i ++ ) { jobManagerRpcServices [ i ] = createRpcService ( configuration , rpcTimeout , true , jobManagerBindAddress ) ; } for ( int i = 0 ; i < numTaskManagers ; i ++ ) { taskManagerRpcServices [ i ] = createRpcService ( configuration , rpcTimeout , true , taskManagerBindAddress ) ; } for ( int i = 0 ; i < numResourceManagers ; i ++ ) { resourceManagerRpcServices [ i ] = createRpcService ( configuration , rpcTimeout , true , resourceManagerBindAddress ) ; } this . jobManagerRpcServices = jobManagerRpcServices ; this . taskManagerRpcServices = taskManagerRpcServices ; this . resourceManagerRpcServices = resourceManagerRpcServices ; } LOG . info ( "Starting high-availability services" ) ; haServices = HighAvailabilityServicesUtils . createAvailableOrEmbeddedServices ( configuration , commonRpcService . getExecutor ( ) ) ; blobServer = new BlobServer ( configuration , haServices . createBlobStore ( ) ) ; blobServer . start ( ) ; heartbeatServices = HeartbeatServices . fromConfiguration ( configuration ) ; LOG . info ( "Starting {} ResourceManger(s)" , numResourceManagers ) ; resourceManagerRunners = startResourceManagers ( configuration , haServices , heartbeatServices , metricRegistry , numResourceManagers , resourceManagerRpcServices ) ; LOG . info ( "Starting {} TaskManger(s)" , numTaskManagers ) ; taskManagers = startTaskManagers ( configuration , haServices , metricRegistry , numTaskManagers , taskManagerRpcServices ) ; LOG . info ( "Starting job dispatcher(s) for {} JobManger(s)" , numJobManagers ) ; jobDispatcher = new MiniClusterJobDispatcher ( configuration , haServices , blobServer , heartbeatServices , metricRegistry , numJobManagers , jobManagerRpcServices ) ; } catch ( Exception e ) { try { shutdownInternally ( ) ; } catch ( Exception ee ) { e . addSuppressed ( ee ) ; } throw e ; } running = true ; LOG . info ( "Flink Mini Cluster started successfully" ) ; } }
Starts the mini cluster, based on the configured properties.
Starts the mini cluster based on the configured properties .
// bring up all the RPC services // we always need the 'commonRpcService' for auxiliary calls // TODO: Temporary hack until the metric query service is ported to the RpcEndpoint // set that same RPC service for all JobManagers and TaskManagers // start a new service per component, possibly with custom bind addresses // create the high-availability services // bring up the ResourceManager(s) // bring up the TaskManager(s) for the mini cluster // bring up the dispatcher that launches JobManagers when jobs submitted // cleanup everything // now officially mark this as running