org.apache.hadoop.io.DataInputByteBuffer#readInt ( )源码实例Demo

/**
 * A helper function to deserialize the metadata returned by ShuffleHandler.
 * @param meta the metadata returned by the ShuffleHandler
 * @return the port the Shuffle Handler is listening on to serve shuffle data.
 */
public static int deserializeMetaData(ByteBuffer meta) throws IOException {
  //TODO this should be returning a class not just an int
  DataInputByteBuffer in = new DataInputByteBuffer();
  in.reset(meta);
  int port = in.readInt();
  return port;
}
 

/**
 * A helper function to deserialize the metadata returned by ShuffleHandler.
 * @param meta the metadata returned by the ShuffleHandler
 * @return the port the Shuffle Handler is listening on to serve shuffle data.
 */
public static int deserializeMetaData(ByteBuffer meta) throws IOException {
  //TODO this should be returning a class not just an int
  DataInputByteBuffer in = new DataInputByteBuffer();
  in.reset(meta);
  int port = in.readInt();
  return port;
}
 

/**
 * A helper function to deserialize the metadata returned by PullServerAuxService.
 * @param meta the metadata returned by the PullServerAuxService
 * @return the port the PullServer Handler is listening on to serve shuffle data.
 */
public static int deserializeMetaData(ByteBuffer meta) throws IOException {
  //TODO this should be returning a class not just an int
  DataInputByteBuffer in = new DataInputByteBuffer();
  in.reset(meta);
  return in.readInt();
}
 

/**
 * A helper function to deserialize the metadata returned by PullServerAuxService.
 * @param meta the metadata returned by the PullServerAuxService
 * @return the port the PullServer Handler is listening on to serve shuffle data.
 */
public static int deserializeMetaData(ByteBuffer meta) throws IOException {
  //TODO this should be returning a class not just an int
  DataInputByteBuffer in = new DataInputByteBuffer();
  in.reset(meta);
  return in.readInt();
}
 

public static int deserializeShuffleProviderMetaData(ByteBuffer meta)
    throws IOException {
  DataInputByteBuffer in = new DataInputByteBuffer();
  try {
    in.reset(meta);
    int port = in.readInt();
    return port;
  } finally {
    in.close();
  }
}
 

public static int deserializeShuffleProviderMetaData(ByteBuffer meta)
    throws IOException {
  DataInputByteBuffer in = new DataInputByteBuffer();
  try {
    in.reset(meta);
    int port = in.readInt();
    return port;
  } finally {
    in.close();
  }
}
 

/**
 * A helper function to deserialize the metadata returned by ShuffleHandler.
 * @param meta the metadata returned by the ShuffleHandler
 * @return the port the Shuffle Handler is listening on to serve shuffle data.
 */
public static int deserializeMetaData(ByteBuffer meta) throws IOException {
  //TODO this should be returning a class not just an int
  DataInputByteBuffer in = new DataInputByteBuffer();
  in.reset(meta);
  int port = in.readInt();
  return port;
}
 

@SuppressWarnings("unchecked")
public synchronized void launch(ContainerLaunchRequest event) {
  LOG.info("Launching " + event.getContainerId());
  if(this.state == ContainerState.KILLED_BEFORE_LAUNCH) {
    state = ContainerState.DONE;
    sendContainerLaunchFailedMsg(event.getContainerId(),
        "Container was killed before it was launched");
    return;
  }

  ContainerManagementProtocolProxyData proxy = null;
  try {

    proxy = getCMProxy(containerID, containerMgrAddress,
        containerToken);

    // Construct the actual Container
    ContainerLaunchContext containerLaunchContext =
      event.getContainerLaunchContext();

    // Now launch the actual container
    StartContainerRequest startRequest = Records
      .newRecord(StartContainerRequest.class);
    startRequest.setContainerToken(event.getContainerToken());
    startRequest.setContainerLaunchContext(containerLaunchContext);

    StartContainersResponse response =
        proxy.getContainerManagementProtocol().startContainers(
            StartContainersRequest.newInstance(
                Collections.singletonList(startRequest)));
    if (response.getFailedRequests() != null
        && !response.getFailedRequests().isEmpty()) {
      throw response.getFailedRequests().get(containerID).deSerialize();
    }

    // after launching, send launched event to task attempt to move
    // it from ASSIGNED to RUNNING state
    getContext().containerLaunched(containerID);
    this.state = ContainerState.RUNNING;

    int shufflePort = TezRuntimeUtils.INVALID_PORT;
    Map<String, java.nio.ByteBuffer> servicesMetaData = response.getAllServicesMetaData();
    if (servicesMetaData != null) {
      String auxiliaryService = conf.get(TezConfiguration.TEZ_AM_SHUFFLE_AUXILIARY_SERVICE_ID,
          TezConfiguration.TEZ_AM_SHUFFLE_AUXILIARY_SERVICE_ID_DEFAULT);
      ByteBuffer portInfo = servicesMetaData.get(auxiliaryService);
      if (portInfo != null) {
        DataInputByteBuffer in = new DataInputByteBuffer();
        in.reset(portInfo);
        shufflePort = in.readInt();
      } else {
        LOG.warn("Shuffle port for {} is not present is the services metadata response", auxiliaryService);
      }
    } else {
      LOG.warn("Shuffle port cannot be found since services metadata response is missing");
    }
    if (deletionTracker != null) {
      deletionTracker.addNodeShufflePort(event.getNodeId(), shufflePort);
    }
  } catch (Throwable t) {
    String message = "Container launch failed for " + containerID + " : "
        + ExceptionUtils.getStackTrace(t);
    this.state = ContainerState.FAILED;
    sendContainerLaunchFailedMsg(containerID, message);
  } finally {
    if (proxy != null) {
      cmProxy.mayBeCloseProxy(proxy);
    }
  }
}