import org.apache.mesos.Protos.Filters; //导入依赖的package包/类
@Override
public Status launchTasks(final OfferID offerId, final Collection<TaskInfo> tasks)
{
    checkNotNull(offerId, "offerId is null");
    checkNotNull(tasks, "tasks is null");

    if (!context.isStateMachine(DRIVER_RUNNING)) {
        return context.getStateMachine();
    }

    final LaunchTasksMessage message = LaunchTasksMessage.newBuilder()
        .setFrameworkId(context.getFrameworkId())
        .addOfferIds(offerId)
        .addAllTasks(tasks)
        .setFilters(Filters.newBuilder().build())
        .build();

    doLaunchTasks(message);
    return context.getStateMachine();
}
 

import org.apache.mesos.Protos.Filters; //导入依赖的package包/类
@Override
public Status declineOffer(final OfferID offerId, final Filters filters)
{
    checkNotNull(offerId, "offerId is null");
    checkNotNull(filters, "filters is null");

    if (!context.isStateMachine(DRIVER_RUNNING)) {
        return context.getStateMachine();
    }

    final LaunchTasksMessage message = LaunchTasksMessage.newBuilder()
        .setFrameworkId(context.getFrameworkId())
        .addOfferIds(offerId)
        .setFilters(filters)
        .build();

    doLaunchTasks(message);
    return context.getStateMachine();
}
 

import org.apache.mesos.Protos.Filters; //导入依赖的package包/类
@Override
public Status launchTasks(
    Collection<OfferID> offerIds,
    Collection<TaskInfo> tasks,
    Filters filters) {

  throw new UnsupportedOperationException();
}
 

import org.apache.mesos.Protos.Filters; //导入依赖的package包/类
@Override
public Status acceptOffers(
    Collection<OfferID> offerIds,
    Collection<Offer.Operation> operations,
    Filters filters) {
  throw new UnsupportedOperationException();
}
 

import org.apache.mesos.Protos.Filters; //导入依赖的package包/类
@Override
public Status declineOffer(OfferID offerId, Filters filters) {
  assertNotStopped();

  throw new UnsupportedOperationException();
}
 

import org.apache.mesos.Protos.Filters; //导入依赖的package包/类
@Override
public Status launchTasks(Collection<OfferID> offerIds, Collection<TaskInfo> tasks, Filters filters) {
  return null;
}
 

import org.apache.mesos.Protos.Filters; //导入依赖的package包/类
@Override
public Status acceptOffers(Collection<OfferID> offerIds, Collection<Operation> operations, Filters filters) {
  return null;
}
 

import org.apache.mesos.Protos.Filters; //导入依赖的package包/类
@Override
public Status declineOffer(OfferID offerId, Filters filters) {
  return null;
}
 

import org.apache.mesos.Protos.Filters; //导入依赖的package包/类
/**
 * Greedily acquire resources by launching a Mesos Task(w/ our custom MesosExecutor) on REEF Evaluator request.
 * Either called from onResourceRequest(for a new request) or resourceOffers(for an outstanding request).
 * TODO[JIRA REEF-102]: reflect priority and rack/node locality specified in resourceRequestEvent.
 */
private synchronized void doResourceRequest(final ResourceRequestEvent resourceRequestEvent) {
  int tasksToLaunchCounter = resourceRequestEvent.getResourceCount();

  for (final Offer offer : this.offers.values()) {
    final int cpuSlots = getCpu(offer) / resourceRequestEvent.getVirtualCores().get();
    final int memSlots = getMemory(offer) / resourceRequestEvent.getMemorySize().get();
    final int taskNum = Math.min(Math.min(cpuSlots, memSlots), tasksToLaunchCounter);

    if (taskNum > 0 && satisfySlaveConstraint(resourceRequestEvent, offer)) {
      final List<TaskInfo> tasksToLaunch = new ArrayList<>();
      tasksToLaunchCounter -= taskNum;

      // Launch as many MesosTasks on the same node(offer) as possible to exploit locality.
      for (int j = 0; j < taskNum; j++) {
        final String id = offer.getId().getValue() + "-" + String.valueOf(j);
        final String executorLaunchCommand = getExecutorLaunchCommand(id, resourceRequestEvent.getMemorySize().get());

        final ExecutorInfo executorInfo = ExecutorInfo.newBuilder()
            .setExecutorId(ExecutorID.newBuilder()
                .setValue(id)
                .build())
            .setCommand(CommandInfo.newBuilder()
                .setValue(executorLaunchCommand)
                .addUris(URI.newBuilder().setValue(reefTarUri).build())
                .build())
            .build();

        final TaskInfo taskInfo = TaskInfo.newBuilder()
            .setTaskId(TaskID.newBuilder()
                .setValue(id)
                .build())
            .setName(id)
            .setSlaveId(offer.getSlaveId())
            .addResources(Resource.newBuilder()
                    .setName("mem")
                    .setType(Type.SCALAR)
                    .setScalar(Value.Scalar.newBuilder()
                            .setValue(resourceRequestEvent.getMemorySize().get())
                            .build())
                    .build())
            .addResources(Resource.newBuilder()
                    .setName("cpus")
                    .setType(Type.SCALAR)
                    .setScalar(Value.Scalar.newBuilder()
                            .setValue(resourceRequestEvent.getVirtualCores().get())
                            .build())
                    .build())
            .setExecutor(executorInfo)
            .build();

        tasksToLaunch.add(taskInfo);
        this.executorIdToLaunchedRequests.put(id, resourceRequestEvent);
      }

      final Filters filters = Filters.newBuilder().setRefuseSeconds(0).build();
      mesosMaster.launchTasks(Collections.singleton(offer.getId()), tasksToLaunch, filters);
    } else {
      mesosMaster.declineOffer(offer.getId());
    }
  }

  // the offers are no longer valid(all launched or declined)
  this.offers.clear();

  // Save leftovers that couldn't be launched
  outstandingRequests.add(ResourceRequestEventImpl.newBuilder()
      .mergeFrom(resourceRequestEvent)
      .setResourceCount(tasksToLaunchCounter)
      .build());
}
 

import org.apache.mesos.Protos.Filters; //导入依赖的package包/类
private void createMesosTask(Offer offer, Request request) {
  final String slaveName = request.request.slave.name;
  TaskID taskId = TaskID.newBuilder().setValue(slaveName).build();

  LOGGER.fine("Launching task " + taskId.getValue() + " with URI " +
              joinPaths(jenkinsMaster, SLAVE_JAR_URI_SUFFIX));

  if (isExistingTask(taskId)) {
      refuseOffer(offer);
      return;
  }

  Jenkins jenkins = getJenkins();
  for (final Computer computer : jenkins.getComputers()) {
      if (!MesosComputer.class.isInstance(computer)) {
          LOGGER.finer("Not a mesos computer, skipping");
          continue;
      }

      MesosComputer mesosComputer = (MesosComputer) computer;
      MesosSlave mesosSlave = mesosComputer.getNode();

      if (taskId.getValue().equals(computer.getName()) && mesosSlave.isPendingDelete()) {
          LOGGER.info("This mesos task " + taskId.getValue() + " is pending deletion. Not launching another task");
          driver.declineOffer(offer.getId());
      }
  }

  CommandInfo.Builder commandBuilder = getCommandInfoBuilder(request);
  TaskInfo.Builder taskBuilder = getTaskInfoBuilder(offer, request, taskId, commandBuilder);

  if (request.request.slaveInfo.getContainerInfo() != null) {
      getContainerInfoBuilder(offer, request, slaveName, taskBuilder);
  }

  List<TaskInfo> tasks = new ArrayList<TaskInfo>();
  tasks.add(taskBuilder.build());
  Filters filters = Filters.newBuilder().setRefuseSeconds(1).build();
  driver.launchTasks(offer.getId(), tasks, filters);

  results.put(taskId, new Result(request.result, new Mesos.JenkinsSlave(offer.getSlaveId()
      .getValue())));
  finishedTasks.add(taskId);
}
 

import org.apache.mesos.Protos.Filters; //导入依赖的package包/类
/**
 * Launches a new Kibana task for the given elasticSearchUrl, using the offer
 *
 * @param requirement the requirement for the task
 * @param offer       the offer used to run the task
 * @param driver      the driver used to launch the task
 */
private void launchNewTask(Map.Entry<String, Integer> requirement, Offer offer, SchedulerDriver driver) {
    TaskInfo task = TaskInfoFactory.buildTask(requirement, offer, configuration);
    configuration.registerTask(requirement.getKey(), task.getTaskId());
    Filters filters = Filters.newBuilder().setRefuseSeconds(1).build();
    driver.launchTasks(Arrays.asList(offer.getId()), Arrays.asList(task), filters);
}