code: public static MesosConfiguration createMesosSchedulerConfiguration(Configuration flinkConfig, String hostname) {
  
  		Protos.FrameworkInfo.Builder frameworkInfo = Protos.FrameworkInfo.newBuilder()
  			.setHostname(hostname);
  		Protos.Credential.Builder credential = null;
  
  		if (!flinkConfig.contains(MesosOptions.MASTER_URL)) {
  			throw new IllegalConfigurationException(MesosOptions.MASTER_URL.key() + " must be configured.");
  		}
  		String masterUrl = flinkConfig.getString(MesosOptions.MASTER_URL);
  
  		Duration failoverTimeout = FiniteDuration.apply(
  			flinkConfig.getInteger(
  				MesosOptions.FAILOVER_TIMEOUT_SECONDS),
  				TimeUnit.SECONDS);
  		frameworkInfo.setFailoverTimeout(failoverTimeout.toSeconds());
  
  		frameworkInfo.setName(flinkConfig.getString(
  			MesosOptions.RESOURCEMANAGER_FRAMEWORK_NAME));
  
  		frameworkInfo.setRole(flinkConfig.getString(
  			MesosOptions.RESOURCEMANAGER_FRAMEWORK_ROLE));
  
  		frameworkInfo.setUser(flinkConfig.getString(
  			MesosOptions.RESOURCEMANAGER_FRAMEWORK_USER));
  
  		if (flinkConfig.contains(MesosOptions.RESOURCEMANAGER_FRAMEWORK_PRINCIPAL)) {
  			frameworkInfo.setPrincipal(flinkConfig.getString(
  				MesosOptions.RESOURCEMANAGER_FRAMEWORK_PRINCIPAL));
  
  			credential = Protos.Credential.newBuilder();
  			credential.setPrincipal(frameworkInfo.getPrincipal());
  
  			// some environments use a side-channel to communicate the secret to Mesos,
  			// and thus don't set the 'secret' configuration setting
  			if (flinkConfig.contains(MesosOptions.RESOURCEMANAGER_FRAMEWORK_SECRET)) {
  				credential.setSecret(flinkConfig.getString(
  					MesosOptions.RESOURCEMANAGER_FRAMEWORK_SECRET));
  			}
  		}
  
  		MesosConfiguration mesos =
  			new MesosConfiguration(masterUrl, frameworkInfo, scala.Option.apply(credential));
  
  		return mesos;
  	}
docs: Loads and validates the Mesos scheduler configuration.
  @param flinkConfig the global configuration.
  @param hostname the hostname to advertise to the Mesos master.