flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosTaskManagerParameters.java
/**
	 * Create the Mesos TaskManager parameters.
	 *
	 * @param flinkConfig the TM configuration.
	 */
Create the Mesos TaskManager parameters.
@param flinkConfig the TM configuration.
Create the Mesos TaskManager parameters . @param flinkConfig the TM configuration .
public static MesosTaskManagerParameters create(Configuration flinkConfig) {
		List<ConstraintEvaluator> constraints = parseConstraints(flinkConfig.getString(MESOS_CONSTRAINTS_HARD_HOSTATTR));
		ContaineredTaskManagerParameters containeredParameters = ContaineredTaskManagerParameters.create(
			flinkConfig,
			flinkConfig.getInteger(MESOS_RM_TASKS_MEMORY_MB),
			flinkConfig.getInteger(MESOS_RM_TASKS_SLOTS));
		double cpus = flinkConfig.getDouble(MESOS_RM_TASKS_CPUS);
		if (cpus <= 0.0) {
			cpus = Math.max(containeredParameters.numSlots(), 1.0);
		}
		String imageName = flinkConfig.getString(MESOS_RM_CONTAINER_IMAGE_NAME);
		ContainerType containerType;
		String containerTypeString = flinkConfig.getString(MESOS_RM_CONTAINER_TYPE);
		switch (containerTypeString) {
			case MESOS_RESOURCEMANAGER_TASKS_CONTAINER_TYPE_MESOS:
				containerType = ContainerType.MESOS;
				break;
			case MESOS_RESOURCEMANAGER_TASKS_CONTAINER_TYPE_DOCKER:
				containerType = ContainerType.DOCKER;
				if (imageName == null || imageName.length() == 0) {
					throw new IllegalConfigurationException(MESOS_RM_CONTAINER_IMAGE_NAME.key() +
						" must be specified for docker container type");
				}
				break;
			default:
				throw new IllegalConfigurationException("invalid container type: " + containerTypeString);
		}
		Option<String> containerVolOpt = Option.<String>apply(flinkConfig.getString(MESOS_RM_CONTAINER_VOLUMES));
		List<Protos.Volume> containerVolumes = buildVolumes(containerVolOpt);
		Option<String> taskManagerHostname = Option.apply(flinkConfig.getString(MESOS_TM_HOSTNAME));
		String tmCommand = flinkConfig.getString(MESOS_TM_CMD);
		Option<String> tmBootstrapCommand = Option.apply(flinkConfig.getString(MESOS_TM_BOOTSTRAP_CMD));
		return new MesosTaskManagerParameters(
			cpus,
			containerType,
			Option.apply(imageName),
			containeredParameters,
			containerVolumes,
			constraints,
			tmCommand,
			tmBootstrapCommand,
			taskManagerHostname);
	}
public static MesosTaskManagerParameters create ( Configuration flinkConfig ) { List < ConstraintEvaluator > constraints = parseConstraints ( flinkConfig . getString ( MESOS_CONSTRAINTS_HARD_HOSTATTR ) ) ; ContaineredTaskManagerParameters containeredParameters = ContaineredTaskManagerParameters . create ( flinkConfig , flinkConfig . getInteger ( MESOS_RM_TASKS_MEMORY_MB ) , flinkConfig . getInteger ( MESOS_RM_TASKS_SLOTS ) ) ; double cpus = flinkConfig . getDouble ( MESOS_RM_TASKS_CPUS ) ; if ( cpus <= 0.0 ) { cpus = Math . max ( containeredParameters . numSlots ( ) , 1.0 ) ; } String imageName = flinkConfig . getString ( MESOS_RM_CONTAINER_IMAGE_NAME ) ; ContainerType containerType ; String containerTypeString = flinkConfig . getString ( MESOS_RM_CONTAINER_TYPE ) ; switch ( containerTypeString ) { case MESOS_RESOURCEMANAGER_TASKS_CONTAINER_TYPE_MESOS : containerType = ContainerType . MESOS ; break ; case MESOS_RESOURCEMANAGER_TASKS_CONTAINER_TYPE_DOCKER : containerType = ContainerType . DOCKER ; if ( imageName == null || imageName . length ( ) == 0 ) { throw new IllegalConfigurationException ( MESOS_RM_CONTAINER_IMAGE_NAME . key ( ) + " must be specified for docker container type" ) ; } break ; default : throw new IllegalConfigurationException ( "invalid container type: " + containerTypeString ) ; } Option < String > containerVolOpt = Option . < String > apply ( flinkConfig . getString ( MESOS_RM_CONTAINER_VOLUMES ) ) ; List < Protos . Volume > containerVolumes = buildVolumes ( containerVolOpt ) ; Option < String > taskManagerHostname = Option . apply ( flinkConfig . getString ( MESOS_TM_HOSTNAME ) ) ; String tmCommand = flinkConfig . getString ( MESOS_TM_CMD ) ; Option < String > tmBootstrapCommand = Option . apply ( flinkConfig . getString ( MESOS_TM_BOOTSTRAP_CMD ) ) ; return new MesosTaskManagerParameters ( cpus , containerType , Option . apply ( imageName ) , containeredParameters , containerVolumes , constraints , tmCommand , tmBootstrapCommand , taskManagerHostname ) ; }
Create the Mesos TaskManager parameters.
Create the Mesos TaskManager parameters .
// parse the common parameters // parse the containerization parameters //obtain Task Manager Host Name from the configuration //obtain command-line from the configuration