private PrepareItemCraftEvent mockPreCraftingEvent(ItemStack item) {
    Player player = server.addPlayer();

    CraftingInventory inv = Mockito.mock(CraftingInventory.class);
    MutableObject result = new MutableObject(new ItemStack(Material.EMERALD));

    Mockito.doAnswer(invocation -> {
        ItemStack argument = invocation.getArgument(0);
        result.setValue(argument);
        return null;
    }).when(inv).setResult(Mockito.any());

    Mockito.when(inv.getResult()).thenAnswer(invocation -> result.getValue());
    Mockito.when(inv.getContents()).thenReturn(new ItemStack[] { null, null, item, null, null, null, null, null, null });

    InventoryView view = player.openInventory(inv);
    PrepareItemCraftEvent event = new PrepareItemCraftEvent(inv, view, false);

    listener.onPrepareCraft(event);
    return event;
}
 

private Resource assignNodeLocalContainers(Resource clusterResource,
    ResourceRequest nodeLocalResourceRequest, FiCaSchedulerNode node,
    FiCaSchedulerApp application, Priority priority,
    RMContainer reservedContainer, MutableObject allocatedContainer,
    ResourceLimits currentResoureLimits) {
  if (canAssign(application, priority, node, NodeType.NODE_LOCAL, 
      reservedContainer)) {
    return assignContainer(clusterResource, node, application, priority,
        nodeLocalResourceRequest, NodeType.NODE_LOCAL, reservedContainer,
        allocatedContainer, currentResoureLimits);
  }
  
  return Resources.none();
}
 

private Resource assignRackLocalContainers(Resource clusterResource,
    ResourceRequest rackLocalResourceRequest, FiCaSchedulerNode node,
    FiCaSchedulerApp application, Priority priority,
    RMContainer reservedContainer, MutableObject allocatedContainer,
    ResourceLimits currentResoureLimits) {
  if (canAssign(application, priority, node, NodeType.RACK_LOCAL,
      reservedContainer)) {
    return assignContainer(clusterResource, node, application, priority,
        rackLocalResourceRequest, NodeType.RACK_LOCAL, reservedContainer,
        allocatedContainer, currentResoureLimits);
  }
  
  return Resources.none();
}
 

private Resource assignOffSwitchContainers(Resource clusterResource,
    ResourceRequest offSwitchResourceRequest, FiCaSchedulerNode node,
    FiCaSchedulerApp application, Priority priority,
    RMContainer reservedContainer, MutableObject allocatedContainer,
    ResourceLimits currentResoureLimits) {
  if (canAssign(application, priority, node, NodeType.OFF_SWITCH,
      reservedContainer)) {
    return assignContainer(clusterResource, node, application, priority,
        offSwitchResourceRequest, NodeType.OFF_SWITCH, reservedContainer,
        allocatedContainer, currentResoureLimits);
  }
  
  return Resources.none();
}
 

private Resource assignNodeLocalContainers(Resource clusterResource,
    ResourceRequest nodeLocalResourceRequest, FiCaSchedulerNode node,
    FiCaSchedulerApp application, Priority priority,
    RMContainer reservedContainer, MutableObject allocatedContainer,
    ResourceLimits currentResoureLimits) {
  if (canAssign(application, priority, node, NodeType.NODE_LOCAL, 
      reservedContainer)) {
    return assignContainer(clusterResource, node, application, priority,
        nodeLocalResourceRequest, NodeType.NODE_LOCAL, reservedContainer,
        allocatedContainer, currentResoureLimits);
  }
  
  return Resources.none();
}
 

private Resource assignRackLocalContainers(Resource clusterResource,
    ResourceRequest rackLocalResourceRequest, FiCaSchedulerNode node,
    FiCaSchedulerApp application, Priority priority,
    RMContainer reservedContainer, MutableObject allocatedContainer,
    ResourceLimits currentResoureLimits) {
  if (canAssign(application, priority, node, NodeType.RACK_LOCAL,
      reservedContainer)) {
    return assignContainer(clusterResource, node, application, priority,
        rackLocalResourceRequest, NodeType.RACK_LOCAL, reservedContainer,
        allocatedContainer, currentResoureLimits);
  }
  
  return Resources.none();
}
 

private Resource assignOffSwitchContainers(Resource clusterResource,
    ResourceRequest offSwitchResourceRequest, FiCaSchedulerNode node,
    FiCaSchedulerApp application, Priority priority,
    RMContainer reservedContainer, MutableObject allocatedContainer,
    ResourceLimits currentResoureLimits) {
  if (canAssign(application, priority, node, NodeType.OFF_SWITCH,
      reservedContainer)) {
    return assignContainer(clusterResource, node, application, priority,
        offSwitchResourceRequest, NodeType.OFF_SWITCH, reservedContainer,
        allocatedContainer, currentResoureLimits);
  }
  
  return Resources.none();
}
 

@Override
public <R> R run(Supplier<R> function) {

    // runBackground synchronously on JavaFX thread
    if (Platform.isFxApplicationThread()) {
        return function.get();
    }

    // queue on JavaFX thread and wait for completion
    final CountDownLatch doneLatch = new CountDownLatch(1);
    final MutableObject result = new MutableObject();
    Platform.runLater(() -> {
        try {
            result.setValue(function.get());
        } finally {
            doneLatch.countDown();
        }
    });

    try {
        doneLatch.await();
    } catch (InterruptedException e) {
        // ignore exception
    }

    return (R) result.getValue();
}
 

void addScheduleGroupCheckers() {
  final GlobalDAGChecker scheduleGroupTopoOrdering = (irdag -> {
    int lastSeenScheduleGroup = Integer.MIN_VALUE;

    for (final IRVertex v : irdag.getVertices()) {
      final MutableObject violatingReachableVertex = new MutableObject();
      v.getPropertyValue(ScheduleGroupProperty.class).ifPresent(startingScheduleGroup ->
        irdag.dfsDo(
          v,
          visited -> {
            if (visited.getPropertyValue(ScheduleGroupProperty.class).isPresent()
              && visited.getPropertyValue(ScheduleGroupProperty.class).get() < startingScheduleGroup) {
              violatingReachableVertex.setValue(visited);
            }
          },
          DAGInterface.TraversalOrder.PreOrder,
          new HashSet<>()));
      if (violatingReachableVertex.getValue() != null) {
        return failure(
          "A reachable vertex with a smaller schedule group ",
          v,
          ScheduleGroupProperty.class,
          violatingReachableVertex.getValue(),
          ScheduleGroupProperty.class);
      }
    }
    return success();
  });
  globalDAGCheckerList.add(scheduleGroupTopoOrdering);

  final SingleEdgeChecker splitByPull = (edge -> {
    if (Util.isControlEdge(edge)) {
      return success();
    }

    if (Optional.of(DataFlowProperty.Value.PULL).equals(edge.getPropertyValue(DataFlowProperty.class))) {
      final Optional<Integer> srcSG = edge.getSrc().getPropertyValue(ScheduleGroupProperty.class);
      final Optional<Integer> dstSG = edge.getDst().getPropertyValue(ScheduleGroupProperty.class);
      if (srcSG.isPresent() && dstSG.isPresent()) {
        if (srcSG.get().equals(dstSG.get())) {
          return failure("Schedule group must split by Pull",
            edge.getSrc(), ScheduleGroupProperty.class, edge.getDst(), ScheduleGroupProperty.class);
        }
      }
    }
    return success();
  });
  singleEdgeCheckerList.add(splitByPull);
}
 

private CSAssignment assignContainersOnNode(Resource clusterResource,
    FiCaSchedulerNode node, FiCaSchedulerApp application, Priority priority,
    RMContainer reservedContainer, ResourceLimits currentResoureLimits) {
  Resource assigned = Resources.none();

  NodeType requestType = null;
  MutableObject allocatedContainer = new MutableObject();
  // Data-local
  ResourceRequest nodeLocalResourceRequest =
      application.getResourceRequest(priority, node.getNodeName());
  if (nodeLocalResourceRequest != null) {
    requestType = NodeType.NODE_LOCAL;
    assigned =
        assignNodeLocalContainers(clusterResource, nodeLocalResourceRequest, 
          node, application, priority, reservedContainer,
          allocatedContainer, currentResoureLimits);
    if (Resources.greaterThan(resourceCalculator, clusterResource,
        assigned, Resources.none())) {

      //update locality statistics
      if (allocatedContainer.getValue() != null) {
        application.incNumAllocatedContainers(NodeType.NODE_LOCAL,
          requestType);
      }
      return new CSAssignment(assigned, NodeType.NODE_LOCAL);
    }
  }

  // Rack-local
  ResourceRequest rackLocalResourceRequest =
      application.getResourceRequest(priority, node.getRackName());
  if (rackLocalResourceRequest != null) {
    if (!rackLocalResourceRequest.getRelaxLocality()) {
      return SKIP_ASSIGNMENT;
    }

    if (requestType != NodeType.NODE_LOCAL) {
      requestType = NodeType.RACK_LOCAL;
    }

    assigned = 
        assignRackLocalContainers(clusterResource, rackLocalResourceRequest, 
          node, application, priority, reservedContainer,
          allocatedContainer, currentResoureLimits);
    if (Resources.greaterThan(resourceCalculator, clusterResource,
        assigned, Resources.none())) {

      //update locality statistics
      if (allocatedContainer.getValue() != null) {
        application.incNumAllocatedContainers(NodeType.RACK_LOCAL,
          requestType);
      }
      return new CSAssignment(assigned, NodeType.RACK_LOCAL);
    }
  }
  
  // Off-switch
  ResourceRequest offSwitchResourceRequest =
      application.getResourceRequest(priority, ResourceRequest.ANY);
  if (offSwitchResourceRequest != null) {
    if (!offSwitchResourceRequest.getRelaxLocality()) {
      return SKIP_ASSIGNMENT;
    }
    if (requestType != NodeType.NODE_LOCAL
        && requestType != NodeType.RACK_LOCAL) {
      requestType = NodeType.OFF_SWITCH;
    }

    assigned =
        assignOffSwitchContainers(clusterResource, offSwitchResourceRequest,
          node, application, priority, reservedContainer,
          allocatedContainer, currentResoureLimits);

    // update locality statistics
    if (allocatedContainer.getValue() != null) {
      application.incNumAllocatedContainers(NodeType.OFF_SWITCH, requestType);
    }
    return new CSAssignment(assigned, NodeType.OFF_SWITCH);
  }
  
  return SKIP_ASSIGNMENT;
}
 

private CSAssignment assignContainersOnNode(Resource clusterResource,
    FiCaSchedulerNode node, FiCaSchedulerApp application, Priority priority,
    RMContainer reservedContainer, ResourceLimits currentResoureLimits) {
  Resource assigned = Resources.none();

  NodeType requestType = null;
  MutableObject allocatedContainer = new MutableObject();
  
  //注意这里的优先级，先分配node local的，再分配rack local的最后分配rack off的
  // Data-local
  ResourceRequest nodeLocalResourceRequest =
      application.getResourceRequest(priority, node.getNodeName());
  if (nodeLocalResourceRequest != null) {
    requestType = NodeType.NODE_LOCAL;
    assigned =
        assignNodeLocalContainers(clusterResource, nodeLocalResourceRequest, 
          node, application, priority, reservedContainer,
          allocatedContainer, currentResoureLimits);
    if (Resources.greaterThan(resourceCalculator, clusterResource,
        assigned, Resources.none())) {

      //update locality statistics
      if (allocatedContainer.getValue() != null) {
        application.incNumAllocatedContainers(NodeType.NODE_LOCAL,
          requestType);
      }
      return new CSAssignment(assigned, NodeType.NODE_LOCAL);
    }
  }

  // Rack-local
  ResourceRequest rackLocalResourceRequest =
      application.getResourceRequest(priority, node.getRackName());
  if (rackLocalResourceRequest != null) {
    if (!rackLocalResourceRequest.getRelaxLocality()) {
      return SKIP_ASSIGNMENT;
    }

    if (requestType != NodeType.NODE_LOCAL) {
      requestType = NodeType.RACK_LOCAL;
    }

    assigned = 
        assignRackLocalContainers(clusterResource, rackLocalResourceRequest, 
          node, application, priority, reservedContainer,
          allocatedContainer, currentResoureLimits);
    if (Resources.greaterThan(resourceCalculator, clusterResource,
        assigned, Resources.none())) {

      //update locality statistics
      if (allocatedContainer.getValue() != null) {
        application.incNumAllocatedContainers(NodeType.RACK_LOCAL,
          requestType);
      }
      return new CSAssignment(assigned, NodeType.RACK_LOCAL);
    }
  }
  
  // Off-switch
  ResourceRequest offSwitchResourceRequest =
      application.getResourceRequest(priority, ResourceRequest.ANY);
  if (offSwitchResourceRequest != null) {
    if (!offSwitchResourceRequest.getRelaxLocality()) {
      return SKIP_ASSIGNMENT;
    }
    if (requestType != NodeType.NODE_LOCAL
        && requestType != NodeType.RACK_LOCAL) {
      requestType = NodeType.OFF_SWITCH;
    }

    assigned =
        assignOffSwitchContainers(clusterResource, offSwitchResourceRequest,
          node, application, priority, reservedContainer,
          allocatedContainer, currentResoureLimits);

    // update locality statistics
    if (allocatedContainer.getValue() != null) {
      application.incNumAllocatedContainers(NodeType.OFF_SWITCH, requestType);
    }
    return new CSAssignment(assigned, NodeType.OFF_SWITCH);
  }
  
  return SKIP_ASSIGNMENT;
}
 

private void doTestReplace(String expectedResult, String replaceTemplate, boolean substring) {
    String expectedShortResult = expectedResult.substring(1, expectedResult.length() - 1);
    StrSubstitutor sub = new StrSubstitutor(values);
    
    // replace using String
    assertEquals(expectedResult, sub.replace(replaceTemplate));
    if (substring) {
        assertEquals(expectedShortResult, sub.replace(replaceTemplate, 1, replaceTemplate.length() - 2));
    }
    
    // replace using char[]
    char[] chars = replaceTemplate.toCharArray();
    assertEquals(expectedResult, sub.replace(chars));
    if (substring) {
        assertEquals(expectedShortResult, sub.replace(chars, 1, chars.length - 2));
    }
    
    // replace using StringBuffer
    StringBuffer buf = new StringBuffer(replaceTemplate);
    assertEquals(expectedResult, sub.replace(buf));
    if (substring) {
        assertEquals(expectedShortResult, sub.replace(buf, 1, buf.length() - 2));
    }
    
    // replace using StrBuilder
    StrBuilder bld = new StrBuilder(replaceTemplate);
    assertEquals(expectedResult, sub.replace(bld));
    if (substring) {
        assertEquals(expectedShortResult, sub.replace(bld, 1, bld.length() - 2));
    }
    
    // replace using object
    MutableObject obj = new MutableObject(replaceTemplate);  // toString returns template
    assertEquals(expectedResult, sub.replace(obj));
    
    // replace in StringBuffer
    buf = new StringBuffer(replaceTemplate);
    assertEquals(true, sub.replaceIn(buf));
    assertEquals(expectedResult, buf.toString());
    if (substring) {
        buf = new StringBuffer(replaceTemplate);
        assertEquals(true, sub.replaceIn(buf, 1, buf.length() - 2));
        assertEquals(expectedResult, buf.toString());  // expect full result as remainder is untouched
    }
    
    // replace in StrBuilder
    bld = new StrBuilder(replaceTemplate);
    assertEquals(true, sub.replaceIn(bld));
    assertEquals(expectedResult, bld.toString());
    if (substring) {
        bld = new StrBuilder(replaceTemplate);
        assertEquals(true, sub.replaceIn(bld, 1, bld.length() - 2));
        assertEquals(expectedResult, bld.toString());  // expect full result as remainder is untouched
    }
}
 

private void doTestReplace(String expectedResult, String replaceTemplate, boolean substring) {
    String expectedShortResult = expectedResult.substring(1, expectedResult.length() - 1);
    StrSubstitutor sub = new StrSubstitutor(values);
    
    // replace using String
    assertEquals(expectedResult, sub.replace(replaceTemplate));
    if (substring) {
        assertEquals(expectedShortResult, sub.replace(replaceTemplate, 1, replaceTemplate.length() - 2));
    }
    
    // replace using char[]
    char[] chars = replaceTemplate.toCharArray();
    assertEquals(expectedResult, sub.replace(chars));
    if (substring) {
        assertEquals(expectedShortResult, sub.replace(chars, 1, chars.length - 2));
    }
    
    // replace using StringBuffer
    StringBuffer buf = new StringBuffer(replaceTemplate);
    assertEquals(expectedResult, sub.replace(buf));
    if (substring) {
        assertEquals(expectedShortResult, sub.replace(buf, 1, buf.length() - 2));
    }
    
    // replace using StrBuilder
    StrBuilder bld = new StrBuilder(replaceTemplate);
    assertEquals(expectedResult, sub.replace(bld));
    if (substring) {
        assertEquals(expectedShortResult, sub.replace(bld, 1, bld.length() - 2));
    }
    
    // replace using object
    MutableObject obj = new MutableObject(replaceTemplate);  // toString returns template
    assertEquals(expectedResult, sub.replace(obj));
    
    // replace in StringBuffer
    buf = new StringBuffer(replaceTemplate);
    assertEquals(true, sub.replaceIn(buf));
    assertEquals(expectedResult, buf.toString());
    if (substring) {
        buf = new StringBuffer(replaceTemplate);
        assertEquals(true, sub.replaceIn(buf, 1, buf.length() - 2));
        assertEquals(expectedResult, buf.toString());  // expect full result as remainder is untouched
    }
    
    // replace in StrBuilder
    bld = new StrBuilder(replaceTemplate);
    assertEquals(true, sub.replaceIn(bld));
    assertEquals(expectedResult, bld.toString());
    if (substring) {
        bld = new StrBuilder(replaceTemplate);
        assertEquals(true, sub.replaceIn(bld, 1, bld.length() - 2));
        assertEquals(expectedResult, bld.toString());  // expect full result as remainder is untouched
    }
}
 

@Override
public ProtocolResponse getProtocolOutput(String url, final Metadata metadata) throws Exception {
    Builder rb = new Request.Builder().url(url);
    customRequestHeaders.forEach((k) -> {
        rb.header(k[0], k[1]);
    });

    if (metadata != null) {
        String lastModified = metadata.getFirstValue("last-modified");
        if (StringUtils.isNotBlank(lastModified)) {
            rb.header("If-Modified-Since",
                    HttpHeaders.formatHttpDate(lastModified));
        }

        String ifNoneMatch = metadata.getFirstValue("etag", protocolMDprefix);
        if (StringUtils.isNotBlank(ifNoneMatch)) {
            rb.header("If-None-Match", ifNoneMatch);
        }

        String accept = metadata.getFirstValue("http.accept");
        if (StringUtils.isNotBlank(accept)) {
            rb.header("Accept", accept);
        }

        String acceptLanguage = metadata.getFirstValue("http.accept.language");
        if (StringUtils.isNotBlank(acceptLanguage)) {
            rb.header("Accept-Language", acceptLanguage);
        }

        if (useCookies) {
            addCookiesToRequest(rb, url, metadata);
        }
        
        String postJSONData = metadata.getFirstValue("http.post.json");
        if (StringUtils.isNotBlank(postJSONData)) {
            RequestBody body = RequestBody.create(JSON, postJSONData);
            rb.post(body);
        }
    }

    Request request = rb.build();
    Call call = client.newCall(request);

    try (Response response = call.execute()) {

        byte[] bytes = new byte[] {};

        Metadata responsemetadata = new Metadata();
        Headers headers = response.headers();

        for (int i = 0, size = headers.size(); i < size; i++) {
            String key = headers.name(i);
            String value = headers.value(i);

            if (key.equals(ProtocolResponse.REQUEST_HEADERS_KEY)
                    || key.equals(ProtocolResponse.RESPONSE_HEADERS_KEY)) {
                value = new String(Base64.getDecoder().decode(value));
            }

            responsemetadata.addValue(key.toLowerCase(Locale.ROOT), value);
        }

        MutableObject trimmed = new MutableObject(TrimmedContentReason.NOT_TRIMMED);
        bytes = toByteArray(response.body(), trimmed);
        if (trimmed.getValue() != TrimmedContentReason.NOT_TRIMMED) {
            if (!call.isCanceled()) {
                call.cancel();
            }
            responsemetadata.setValue(ProtocolResponse.TRIMMED_RESPONSE_KEY,
                    "true");
            responsemetadata.setValue(
                    ProtocolResponse.TRIMMED_RESPONSE_REASON_KEY,
                    trimmed.getValue().toString().toLowerCase(Locale.ROOT));
            LOG.warn("HTTP content trimmed to {}", bytes.length);
        }

        return new ProtocolResponse(bytes, response.code(), responsemetadata);
    }
}
 

private final byte[] toByteArray(final ResponseBody responseBody,
        MutableObject trimmed) throws IOException {

    if (responseBody == null) {
        return new byte[] {};
    }

    int maxContentBytes = Integer.MAX_VALUE;
    if (maxContent != -1) {
        maxContentBytes = Math.min(maxContentBytes, maxContent);
    }

    long endDueFor = -1;
    if (completionTimeout != -1) {
        endDueFor = System.currentTimeMillis() + (completionTimeout * 1000);
    }

    BufferedSource source = responseBody.source();
    int bytesRequested = 0;
    int bufferGrowStepBytes = 8192;

    while (source.getBuffer().size() <= maxContentBytes) {
        bytesRequested += Math.min(bufferGrowStepBytes,
                /*
                 * request one byte more than required to reliably detect
                 * truncated content, but beware of integer overflows
                 */
                (maxContentBytes == Integer.MAX_VALUE ? maxContentBytes
                        : (1 + maxContentBytes)) - bytesRequested);
        boolean success = false;
        try {
            success = source.request(bytesRequested);
        } catch (IOException e) {
            // requesting more content failed, e.g. by a socket timeout
            if (partialContentAsTrimmed && source.getBuffer().size() > 0) {
                // treat already fetched content as trimmed
                trimmed.setValue(TrimmedContentReason.DISCONNECT);
                LOG.debug("Exception while fetching {}", e);
            } else {
                throw e;
            }
        }
        if (!success) {
            // source exhausted, no more data to read
            break;
        }

        if (endDueFor != -1 && endDueFor <= System.currentTimeMillis()) {
            // check whether we hit the completion timeout
            trimmed.setValue(TrimmedContentReason.TIME);
            break;
        }

        // okhttp may fetch more content than requested, quickly "increment"
        // bytes
        bytesRequested = (int) source.getBuffer().size();
    }
    int bytesBuffered = (int) source.getBuffer().size();
    int bytesToCopy = bytesBuffered;
    if (maxContent != -1 && bytesToCopy > maxContent) {
        // okhttp's internal buffer is larger than maxContent
        trimmed.setValue(TrimmedContentReason.LENGTH);
        bytesToCopy = maxContentBytes;
    }
    byte[] arr = new byte[bytesToCopy];
    source.getBuffer().readFully(arr);
    return arr;
}