docs: public void start() throws Exception {
  		synchronized (lock) {
  			checkState(!running, "MiniCluster is already running");
  
  			LOG.info("Starting Flink Mini Cluster");
  			LOG.debug("Using configuration {}", miniClusterConfiguration);
  
  			final Configuration configuration = miniClusterConfiguration.getConfiguration();
  			final boolean useSingleRpcService = miniClusterConfiguration.getRpcServiceSharing() == RpcServiceSharing.SHARED;
  
  			try {
  				initializeIOFormatClasses(configuration);
  
  				LOG.info("Starting Metrics Registry");
  				metricRegistry = createMetricRegistry(configuration);
  
  				// bring up all the RPC services
  				LOG.info("Starting RPC Service(s)");
  
  				AkkaRpcServiceConfiguration akkaRpcServiceConfig = AkkaRpcServiceConfiguration.fromConfiguration(configuration);
  
  				final RpcServiceFactory dispatcherResourceManagreComponentRpcServiceFactory;
  
  				if (useSingleRpcService) {
  					// we always need the 'commonRpcService' for auxiliary calls
  					commonRpcService = createRpcService(akkaRpcServiceConfig, false, null);
  					final CommonRpcServiceFactory commonRpcServiceFactory = new CommonRpcServiceFactory(commonRpcService);
  					taskManagerRpcServiceFactory = commonRpcServiceFactory;
  					dispatcherResourceManagreComponentRpcServiceFactory = commonRpcServiceFactory;
  				} else {
  					// we always need the 'commonRpcService' for auxiliary calls
  					commonRpcService = createRpcService(akkaRpcServiceConfig, true, null);
  
  					// start a new service per component, possibly with custom bind addresses
  					final String jobManagerBindAddress = miniClusterConfiguration.getJobManagerBindAddress();
  					final String taskManagerBindAddress = miniClusterConfiguration.getTaskManagerBindAddress();
  
  					dispatcherResourceManagreComponentRpcServiceFactory = new DedicatedRpcServiceFactory(akkaRpcServiceConfig, jobManagerBindAddress);
  					taskManagerRpcServiceFactory = new DedicatedRpcServiceFactory(akkaRpcServiceConfig, taskManagerBindAddress);
  				}
  
  				RpcService metricQueryServiceRpcService = MetricUtils.startMetricsRpcService(
  					configuration,
  					commonRpcService.getAddress());
  				metricRegistry.startQueryService(metricQueryServiceRpcService, null);
  
  				ioExecutor = Executors.newFixedThreadPool(
  					Hardware.getNumberCPUCores(),
  					new ExecutorThreadFactory("mini-cluster-io"));
  				haServices = createHighAvailabilityServices(configuration, ioExecutor);
  
  				blobServer = new BlobServer(configuration, haServices.createBlobStore());
  				blobServer.start();
  
  				heartbeatServices = HeartbeatServices.fromConfiguration(configuration);
  
  				blobCacheService = new BlobCacheService(
  					configuration, haServices.createBlobStore(), new InetSocketAddress(InetAddress.getLocalHost(), blobServer.getPort())
  				);
  
  				startTaskManagers();
  
  				MetricQueryServiceRetriever metricQueryServiceRetriever = new RpcMetricQueryServiceRetriever(metricRegistry.getMetricQueryServiceRpcService());
  
  				dispatcherResourceManagerComponents.addAll(createDispatcherResourceManagerComponents(
  					configuration,
  					dispatcherResourceManagreComponentRpcServiceFactory,
  					haServices,
  					blobServer,
  					heartbeatServices,
  					metricRegistry,
  					metricQueryServiceRetriever,
  					new ShutDownFatalErrorHandler()
  				));
  
  				resourceManagerLeaderRetriever = haServices.getResourceManagerLeaderRetriever();
  				dispatcherLeaderRetriever = haServices.getDispatcherLeaderRetriever();
  				webMonitorLeaderRetrievalService = haServices.getWebMonitorLeaderRetriever();
  
  				dispatcherGatewayRetriever = new RpcGatewayRetriever<>(
  					commonRpcService,
  					DispatcherGateway.class,
  					DispatcherId::fromUuid,
  					20,
  					Time.milliseconds(20L));
  				resourceManagerGatewayRetriever = new RpcGatewayRetriever<>(
  					commonRpcService,
  					ResourceManagerGateway.class,
  					ResourceManagerId::fromUuid,
  					20,
  					Time.milliseconds(20L));
  				webMonitorLeaderRetriever = new LeaderRetriever();
  
  				resourceManagerLeaderRetriever.start(resourceManagerGatewayRetriever);
  				dispatcherLeaderRetriever.start(dispatcherGatewayRetriever);
  				webMonitorLeaderRetrievalService.start(webMonitorLeaderRetriever);
  			}
  			catch (Exception e) {
  				// cleanup everything
  				try {
  					close();
  				} catch (Exception ee) {
  					e.addSuppressed(ee);
  				}
  				throw e;
  			}
  
  			// create a new termination future
  			terminationFuture = new CompletableFuture<>();
  
  			// now officially mark this as running
  			running = true;
  
  			LOG.info("Flink Mini Cluster started successfully");
  		}
  	}