docs: @Override
  	protected void initialize() throws ResourceManagerException {
  		// create and start the worker store
  		try {
  			this.workerStore = mesosServices.createMesosWorkerStore(flinkConfig, getRpcService().getExecutor());
  			workerStore.start();
  		} catch (Exception e) {
  			throw new ResourceManagerException("Unable to initialize the worker store.", e);
  		}
  
  		// Prepare to register with Mesos
  		Protos.FrameworkInfo.Builder frameworkInfo = mesosConfig.frameworkInfo()
  			.clone()
  			.setCheckpoint(true);
  		if (webUiUrl != null) {
  			frameworkInfo.setWebuiUrl(webUiUrl);
  		}
  
  		try {
  			Option<Protos.FrameworkID> frameworkID = workerStore.getFrameworkID();
  			if (frameworkID.isEmpty()) {
  				LOG.info("Registering as new framework.");
  			} else {
  				LOG.info("Recovery scenario: re-registering using framework ID {}.", frameworkID.get().getValue());
  				frameworkInfo.setId(frameworkID.get());
  			}
  		} catch (Exception e) {
  			throw new ResourceManagerException("Unable to recover the framework ID.", e);
  		}
  
  		initializedMesosConfig = mesosConfig.withFrameworkInfo(frameworkInfo);
  		MesosConfiguration.logMesosConfig(LOG, initializedMesosConfig);
  
  		this.selfActor = createSelfActor();
  
  		// configure the artifact server to serve the TM container artifacts
  		try {
  			LaunchableMesosWorker.configureArtifactServer(artifactServer, taskManagerContainerSpec);
  		}
  		catch (IOException e) {
  			throw new ResourceManagerException("Unable to configure the artifact server with TaskManager artifacts.", e);
  		}
  	}