@Test(expected = IllegalConfigurationException.class) public void testNegativeNumberOfGPUs() throws Exception { MesosTaskManagerParameters.create(withGPUConfiguration(-1)); }
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); } int gpus = flinkConfig.getInteger(MESOS_RM_TASKS_GPUS); if (gpus < 0) { throw new IllegalConfigurationException(MESOS_RM_TASKS_GPUS.key() + " cannot be negative"); } int disk = flinkConfig.getInteger(MESOS_RM_TASKS_DISK_MB); 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)); Option<String> dockerParamsOpt = Option.<String>apply(flinkConfig.getString(MESOS_RM_CONTAINER_DOCKER_PARAMETERS)); Option<String> uriParamsOpt = Option.<String>apply(flinkConfig.getString(MESOS_TM_URIS)); boolean dockerForcePullImage = flinkConfig.getBoolean(MESOS_RM_CONTAINER_DOCKER_FORCE_PULL_IMAGE); List<Protos.Volume> containerVolumes = buildVolumes(containerVolOpt); List<Protos.Parameter> dockerParameters = buildDockerParameters(dockerParamsOpt); List<String> uris = buildUris(uriParamsOpt); 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, gpus, disk, containerType, Option.apply(imageName), containeredParameters, containerVolumes, dockerParameters, dockerForcePullImage, constraints, tmCommand, tmBootstrapCommand, taskManagerHostname, uris); }
MesosTaskManagerParameters { 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); } int gpus = flinkConfig.getInteger(MESOS_RM_TASKS_GPUS); if (gpus < 0) { throw new IllegalConfigurationException(MESOS_RM_TASKS_GPUS.key() + " cannot be negative"); } int disk = flinkConfig.getInteger(MESOS_RM_TASKS_DISK_MB); 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)); Option<String> dockerParamsOpt = Option.<String>apply(flinkConfig.getString(MESOS_RM_CONTAINER_DOCKER_PARAMETERS)); Option<String> uriParamsOpt = Option.<String>apply(flinkConfig.getString(MESOS_TM_URIS)); boolean dockerForcePullImage = flinkConfig.getBoolean(MESOS_RM_CONTAINER_DOCKER_FORCE_PULL_IMAGE); List<Protos.Volume> containerVolumes = buildVolumes(containerVolOpt); List<Protos.Parameter> dockerParameters = buildDockerParameters(dockerParamsOpt); List<String> uris = buildUris(uriParamsOpt); 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, gpus, disk, containerType, Option.apply(imageName), containeredParameters, containerVolumes, dockerParameters, dockerForcePullImage, constraints, tmCommand, tmBootstrapCommand, taskManagerHostname, uris); } }
MesosTaskManagerParameters { 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); } int gpus = flinkConfig.getInteger(MESOS_RM_TASKS_GPUS); if (gpus < 0) { throw new IllegalConfigurationException(MESOS_RM_TASKS_GPUS.key() + " cannot be negative"); } int disk = flinkConfig.getInteger(MESOS_RM_TASKS_DISK_MB); 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)); Option<String> dockerParamsOpt = Option.<String>apply(flinkConfig.getString(MESOS_RM_CONTAINER_DOCKER_PARAMETERS)); Option<String> uriParamsOpt = Option.<String>apply(flinkConfig.getString(MESOS_TM_URIS)); boolean dockerForcePullImage = flinkConfig.getBoolean(MESOS_RM_CONTAINER_DOCKER_FORCE_PULL_IMAGE); List<Protos.Volume> containerVolumes = buildVolumes(containerVolOpt); List<Protos.Parameter> dockerParameters = buildDockerParameters(dockerParamsOpt); List<String> uris = buildUris(uriParamsOpt); 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, gpus, disk, containerType, Option.apply(imageName), containeredParameters, containerVolumes, dockerParameters, dockerForcePullImage, constraints, tmCommand, tmBootstrapCommand, taskManagerHostname, uris); } MesosTaskManagerParameters(
			double cpus,
			int gpus,
			int disk,
			ContainerType containerType,
			Option<String> containerImageName,
			ContaineredTaskManagerParameters containeredParameters,
			List<Protos.Volume> containerVolumes,
			List<Protos.Parameter> dockerParameters,
			boolean dockerForcePullImage,
			List<ConstraintEvaluator> constraints,
			String command,
			Option<String> bootstrapCommand,
			Option<String> taskManagerHostname,
			List<String> uris); }
MesosTaskManagerParameters { 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); } int gpus = flinkConfig.getInteger(MESOS_RM_TASKS_GPUS); if (gpus < 0) { throw new IllegalConfigurationException(MESOS_RM_TASKS_GPUS.key() + " cannot be negative"); } int disk = flinkConfig.getInteger(MESOS_RM_TASKS_DISK_MB); 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)); Option<String> dockerParamsOpt = Option.<String>apply(flinkConfig.getString(MESOS_RM_CONTAINER_DOCKER_PARAMETERS)); Option<String> uriParamsOpt = Option.<String>apply(flinkConfig.getString(MESOS_TM_URIS)); boolean dockerForcePullImage = flinkConfig.getBoolean(MESOS_RM_CONTAINER_DOCKER_FORCE_PULL_IMAGE); List<Protos.Volume> containerVolumes = buildVolumes(containerVolOpt); List<Protos.Parameter> dockerParameters = buildDockerParameters(dockerParamsOpt); List<String> uris = buildUris(uriParamsOpt); 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, gpus, disk, containerType, Option.apply(imageName), containeredParameters, containerVolumes, dockerParameters, dockerForcePullImage, constraints, tmCommand, tmBootstrapCommand, taskManagerHostname, uris); } MesosTaskManagerParameters(
			double cpus,
			int gpus,
			int disk,
			ContainerType containerType,
			Option<String> containerImageName,
			ContaineredTaskManagerParameters containeredParameters,
			List<Protos.Volume> containerVolumes,
			List<Protos.Parameter> dockerParameters,
			boolean dockerForcePullImage,
			List<ConstraintEvaluator> constraints,
			String command,
			Option<String> bootstrapCommand,
			Option<String> taskManagerHostname,
			List<String> uris); double cpus(); int gpus(); int disk(); ContainerType containerType(); Option<String> containerImageName(); ContaineredTaskManagerParameters containeredParameters(); List<Protos.Volume> containerVolumes(); List<Protos.Parameter> dockerParameters(); boolean dockerForcePullImage(); List<ConstraintEvaluator> constraints(); Option<String> getTaskManagerHostname(); String command(); Option<String> bootstrapCommand(); List<String> uris(); @Override String toString(); static MesosTaskManagerParameters create(Configuration flinkConfig); static List<Protos.Volume> buildVolumes(Option<String> containerVolumes); static List<Protos.Parameter> buildDockerParameters(Option<String> dockerParameters); static List<String> buildUris(Option<String> uris); }
MesosTaskManagerParameters { 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); } int gpus = flinkConfig.getInteger(MESOS_RM_TASKS_GPUS); if (gpus < 0) { throw new IllegalConfigurationException(MESOS_RM_TASKS_GPUS.key() + " cannot be negative"); } int disk = flinkConfig.getInteger(MESOS_RM_TASKS_DISK_MB); 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)); Option<String> dockerParamsOpt = Option.<String>apply(flinkConfig.getString(MESOS_RM_CONTAINER_DOCKER_PARAMETERS)); Option<String> uriParamsOpt = Option.<String>apply(flinkConfig.getString(MESOS_TM_URIS)); boolean dockerForcePullImage = flinkConfig.getBoolean(MESOS_RM_CONTAINER_DOCKER_FORCE_PULL_IMAGE); List<Protos.Volume> containerVolumes = buildVolumes(containerVolOpt); List<Protos.Parameter> dockerParameters = buildDockerParameters(dockerParamsOpt); List<String> uris = buildUris(uriParamsOpt); 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, gpus, disk, containerType, Option.apply(imageName), containeredParameters, containerVolumes, dockerParameters, dockerForcePullImage, constraints, tmCommand, tmBootstrapCommand, taskManagerHostname, uris); } MesosTaskManagerParameters(
			double cpus,
			int gpus,
			int disk,
			ContainerType containerType,
			Option<String> containerImageName,
			ContaineredTaskManagerParameters containeredParameters,
			List<Protos.Volume> containerVolumes,
			List<Protos.Parameter> dockerParameters,
			boolean dockerForcePullImage,
			List<ConstraintEvaluator> constraints,
			String command,
			Option<String> bootstrapCommand,
			Option<String> taskManagerHostname,
			List<String> uris); double cpus(); int gpus(); int disk(); ContainerType containerType(); Option<String> containerImageName(); ContaineredTaskManagerParameters containeredParameters(); List<Protos.Volume> containerVolumes(); List<Protos.Parameter> dockerParameters(); boolean dockerForcePullImage(); List<ConstraintEvaluator> constraints(); Option<String> getTaskManagerHostname(); String command(); Option<String> bootstrapCommand(); List<String> uris(); @Override String toString(); static MesosTaskManagerParameters create(Configuration flinkConfig); static List<Protos.Volume> buildVolumes(Option<String> containerVolumes); static List<Protos.Parameter> buildDockerParameters(Option<String> dockerParameters); static List<String> buildUris(Option<String> uris); static final Pattern TASK_ID_PATTERN; static final ConfigOption<Integer> MESOS_RM_TASKS_SLOTS; static final ConfigOption<Integer> MESOS_RM_TASKS_MEMORY_MB; static final ConfigOption<Integer> MESOS_RM_TASKS_DISK_MB; static final ConfigOption<Double> MESOS_RM_TASKS_CPUS; static final ConfigOption<Integer> MESOS_RM_TASKS_GPUS; static final ConfigOption<String> MESOS_RM_CONTAINER_TYPE; static final ConfigOption<String> MESOS_RM_CONTAINER_IMAGE_NAME; static final ConfigOption<String> MESOS_TM_HOSTNAME; static final ConfigOption<String> MESOS_TM_CMD; static final ConfigOption<String> MESOS_TM_BOOTSTRAP_CMD; static final ConfigOption<String> MESOS_TM_URIS; static final ConfigOption<String> MESOS_RM_CONTAINER_VOLUMES; static final ConfigOption<String> MESOS_RM_CONTAINER_DOCKER_PARAMETERS; static final ConfigOption<Boolean> MESOS_RM_CONTAINER_DOCKER_FORCE_PULL_IMAGE; static final ConfigOption<String> MESOS_CONSTRAINTS_HARD_HOSTATTR; static final String MESOS_RESOURCEMANAGER_TASKS_CONTAINER_TYPE_MESOS; static final String MESOS_RESOURCEMANAGER_TASKS_CONTAINER_TYPE_DOCKER; }