import org.apache.hadoop.yarn.util.AuxiliaryServiceHelper; //导入依赖的package包/类
@Test (timeout = 5000)
public void testAuxiliaryServiceHelper() throws Exception {
  Map<String, String> env = new HashMap<String, String>();
  String serviceName = "testAuxiliaryService";
  ByteBuffer bb = ByteBuffer.wrap("testAuxiliaryService".getBytes());
  AuxiliaryServiceHelper.setServiceDataIntoEnv(serviceName, bb, env);
  Assert.assertEquals(bb,
      AuxiliaryServiceHelper.getServiceDataFromEnv(serviceName, env));
}
 

import org.apache.hadoop.yarn.util.AuxiliaryServiceHelper; //导入依赖的package包/类
@Nullable
@Override
public ByteBuffer getServiceProviderMetaData(String serviceName) {
  Preconditions.checkNotNull(serviceName, "serviceName is null");
  return AuxiliaryServiceHelper.getServiceDataFromEnv(
      serviceName, auxServiceEnv);
}
 

import org.apache.hadoop.yarn.util.AuxiliaryServiceHelper; //导入依赖的package包/类
@Test
public void testGeneratedDataMovementEvent() throws Exception {

  OnFileUnorderedKVOutput kvOutput = new OnFileUnorderedKVOutputForTest();

  Configuration conf = new Configuration();
  conf.set(TezJobConfig.TEZ_RUNTIME_KEY_CLASS, Text.class.getName());
  conf.set(TezJobConfig.TEZ_RUNTIME_VALUE_CLASS, IntWritable.class.getName());

  int appAttemptNumber = 1;
  TezUmbilical tezUmbilical = null;
  String dagName = "currentDAG";
  String taskVertexName = "currentVertex";
  String destinationVertexName = "destinationVertex";
  TezDAGID dagID = TezDAGID.getInstance("2000", 1, 1);
  TezVertexID vertexID = TezVertexID.getInstance(dagID, 1);
  TezTaskID taskID = TezTaskID.getInstance(vertexID, 1);
  TezTaskAttemptID taskAttemptID = TezTaskAttemptID.getInstance(taskID, 1);
  TezCounters counters = new TezCounters();
  byte[] userPayload = TezUtils.createUserPayloadFromConf(conf);
  RuntimeTask runtimeTask = mock(RuntimeTask.class);
  
  int shufflePort = 2112;
  Map<String, String> auxEnv = new HashMap<String, String>();
  ByteBuffer bb = ByteBuffer.allocate(4);
  bb.putInt(shufflePort);
  bb.position(0);
  AuxiliaryServiceHelper.setServiceDataIntoEnv(ShuffleUtils.SHUFFLE_HANDLER_SERVICE_ID, bb, auxEnv);


  OutputDescriptor outputDescriptor = mock(OutputDescriptor.class);
  when(outputDescriptor.getClassName()).thenReturn("OutputDescriptor");

  TezOutputContext outputContext = new TezOutputContextImpl(conf, new String[] {workDir.toString()},
      appAttemptNumber, tezUmbilical, dagName, taskVertexName, destinationVertexName,
      taskAttemptID, counters, 0, userPayload, runtimeTask,
      null, auxEnv, new MemoryDistributor(1, 1, conf) , outputDescriptor);

  List<Event> events = null;

  events = kvOutput.initialize(outputContext);
  assertTrue(events != null && events.size() == 0);

  KeyValueWriter kvWriter = kvOutput.getWriter();
  List<KVPair> data = KVDataGen.generateTestData(true);
  for (KVPair kvp : data) {
    kvWriter.write(kvp.getKey(), kvp.getvalue());
  }

  events = kvOutput.close();
  assertTrue(events != null && events.size() == 1);
  DataMovementEvent dmEvent = (DataMovementEvent)events.get(0);

  assertEquals("Invalid source index", 0, dmEvent.getSourceIndex());

  DataMovementEventPayloadProto shufflePayload = DataMovementEventPayloadProto
      .parseFrom(dmEvent.getUserPayload());

  assertFalse(shufflePayload.hasEmptyPartitions());
  assertEquals(outputContext.getUniqueIdentifier(), shufflePayload.getPathComponent());
  assertEquals(shufflePort, shufflePayload.getPort());
  assertEquals("host", shufflePayload.getHost());
}
 

import org.apache.hadoop.yarn.util.AuxiliaryServiceHelper; //导入依赖的package包/类
public void setShufflePort(String auxiliaryService, int shufflePort) {
  AuxiliaryServiceHelper.setServiceDataIntoEnv(
      auxiliaryService,
      ByteBuffer.allocate(4).putInt(shufflePort), localEnv);
}
 

import org.apache.hadoop.yarn.util.AuxiliaryServiceHelper; //导入依赖的package包/类
private OutputContext createOutputContext(Configuration conf, TezSharedExecutor sharedExecutor)
    throws IOException {
  int appAttemptNumber = 1;
  TezUmbilical tezUmbilical = mock(TezUmbilical.class);
  String dagName = "currentDAG";
  String taskVertexName = "currentVertex";
  String destinationVertexName = "destinationVertex";
  TezDAGID dagID = TezDAGID.getInstance("2000", 1, 1);
  TezVertexID vertexID = TezVertexID.getInstance(dagID, 1);
  TezTaskID taskID = TezTaskID.getInstance(vertexID, 1);
  TezTaskAttemptID taskAttemptID = TezTaskAttemptID.getInstance(taskID, 1);
  UserPayload userPayload = TezUtils.createUserPayloadFromConf(conf);
  
  TaskSpec mockSpec = mock(TaskSpec.class);
  when(mockSpec.getInputs()).thenReturn(Collections.singletonList(mock(InputSpec.class)));
  when(mockSpec.getOutputs()).thenReturn(Collections.singletonList(mock(OutputSpec.class)));
  task = new LogicalIOProcessorRuntimeTask(mockSpec, appAttemptNumber, new Configuration(),
      new String[]{"/"}, tezUmbilical, null, null, null, null, "", null, 1024, false,
      new DefaultHadoopShim(), sharedExecutor);

  LogicalIOProcessorRuntimeTask runtimeTask = spy(task);
  
  Map<String, String> auxEnv = new HashMap<String, String>();
  ByteBuffer bb = ByteBuffer.allocate(4);
  bb.putInt(shufflePort);
  bb.position(0);
  AuxiliaryServiceHelper.setServiceDataIntoEnv(conf.get(TezConfiguration.TEZ_AM_SHUFFLE_AUXILIARY_SERVICE_ID,
      TezConfiguration.TEZ_AM_SHUFFLE_AUXILIARY_SERVICE_ID_DEFAULT), bb, auxEnv);


  OutputDescriptor outputDescriptor = mock(OutputDescriptor.class);
  when(outputDescriptor.getClassName()).thenReturn("OutputDescriptor");

  OutputContext realOutputContext = new TezOutputContextImpl(conf, new String[] {workDir.toString()},
      appAttemptNumber, tezUmbilical, dagName, taskVertexName, destinationVertexName,
      -1, taskAttemptID, 0, userPayload, runtimeTask,
      null, auxEnv, new MemoryDistributor(1, 1, conf) , outputDescriptor, null,
      new ExecutionContextImpl("localhost"), 2048, new TezSharedExecutor(defaultConf));
  verify(runtimeTask, times(1)).addAndGetTezCounter(destinationVertexName);
  verify(runtimeTask, times(1)).getTaskStatistics();
  // verify output stats object got created
  Assert.assertTrue(task.getTaskStatistics().getIOStatistics().containsKey(destinationVertexName));
  OutputContext outputContext = spy(realOutputContext);
  doAnswer(new Answer() {
    @Override public Object answer(InvocationOnMock invocation) throws Throwable {
      long requestedSize = (Long) invocation.getArguments()[0];
      MemoryUpdateCallbackHandler callback = (MemoryUpdateCallbackHandler) invocation
          .getArguments()[1];
      callback.memoryAssigned(requestedSize);
      return null;
    }
  }).when(outputContext).requestInitialMemory(anyLong(), any(MemoryUpdateCallback.class));

  return outputContext;
}
 

import org.apache.hadoop.yarn.util.AuxiliaryServiceHelper; //导入依赖的package包/类
public static LogicalIOProcessorRuntimeTask createLogicalTask(FileSystem fs, Path workDir,
    JobConf jobConf, int mapId, Path mapInput,
    TezUmbilical umbilical, String dagName,
    String vertexName, List<InputSpec> inputSpecs,
    List<OutputSpec> outputSpecs, TezSharedExecutor sharedExecutor) throws Exception {
  jobConf.setInputFormat(SequenceFileInputFormat.class);

  ProcessorDescriptor mapProcessorDesc = ProcessorDescriptor.create(
      MapProcessor.class.getName()).setUserPayload(
      TezUtils.createUserPayloadFromConf(jobConf));
  
  Token<JobTokenIdentifier> shuffleToken = new Token<JobTokenIdentifier>();

  TaskSpec taskSpec = new TaskSpec(
      TezTestUtils.getMockTaskAttemptId(0, 0, mapId, 0),
      dagName, vertexName, -1,
      mapProcessorDesc,
      inputSpecs,
      outputSpecs, null, null);

  Map<String, ByteBuffer> serviceConsumerMetadata = new HashMap<String, ByteBuffer>();
  String auxiliaryService = jobConf.get(TezConfiguration.TEZ_AM_SHUFFLE_AUXILIARY_SERVICE_ID,
      TezConfiguration.TEZ_AM_SHUFFLE_AUXILIARY_SERVICE_ID_DEFAULT);
  serviceConsumerMetadata.put(auxiliaryService,
      ShuffleUtils.convertJobTokenToBytes(shuffleToken));
  Map<String, String> envMap = new HashMap<String, String>();
  ByteBuffer shufflePortBb = ByteBuffer.allocate(4).putInt(0, 8000);
  AuxiliaryServiceHelper
      .setServiceDataIntoEnv(auxiliaryService, shufflePortBb,
          envMap);

  LogicalIOProcessorRuntimeTask task = new LogicalIOProcessorRuntimeTask(
      taskSpec,
      0,
      jobConf,
      new String[] {workDir.toString()},
      umbilical,
      serviceConsumerMetadata,
      envMap,
      HashMultimap.<String, String>create(), null, "", new ExecutionContextImpl("localhost"),
      Runtime.getRuntime().maxMemory(), true, new DefaultHadoopShim(), sharedExecutor);
  return task;
}
 

import org.apache.hadoop.yarn.util.AuxiliaryServiceHelper; //导入依赖的package包/类
public LocalContainerLauncher(ContainerLauncherContext containerLauncherContext,
                              AppContext context,
                              TaskCommunicatorManagerInterface taskCommunicatorManagerInterface,
                              String workingDirectory,
                              boolean isLocalMode) throws UnknownHostException, TezException {
  // TODO Post TEZ-2003. Most of this information is dynamic and only available after the AM
  // starts up. It's not possible to set these up via a static payload.
  // Will need some kind of mechanism to dynamically crate payloads / bind to parameters
  // after the AM starts up.
  super(containerLauncherContext);
  this.context = context;
  this.tal = taskCommunicatorManagerInterface;
  this.workingDirectory = workingDirectory;
  this.isLocalMode = isLocalMode;

  // Check if the hostname is set in the environment before overriding it.
  String host = isLocalMode ? InetAddress.getLocalHost().getHostName() :
      System.getenv(Environment.NM_HOST.name());
  executionContext = new ExecutionContextImpl(host);

  Configuration conf;
  try {
    conf = TezUtils.createConfFromUserPayload(getContext().getInitialUserPayload());
  } catch (IOException e) {
    throw new TezUncheckedException(
        "Failed to parse user payload for " + LocalContainerLauncher.class.getSimpleName(), e);
  }
  if (isLocalMode) {
    String auxiliaryService = conf.get(TezConfiguration.TEZ_AM_SHUFFLE_AUXILIARY_SERVICE_ID,
        TezConfiguration.TEZ_AM_SHUFFLE_AUXILIARY_SERVICE_ID_DEFAULT);
    localEnv = Maps.newHashMap();
    shufflePort = 0;
    AuxiliaryServiceHelper.setServiceDataIntoEnv(
        auxiliaryService, ByteBuffer.allocate(4).putInt(shufflePort), localEnv);
  } else {
    localEnv = System.getenv();
  }
  numExecutors = conf.getInt(TezConfiguration.TEZ_AM_INLINE_TASK_EXECUTION_MAX_TASKS,
      TezConfiguration.TEZ_AM_INLINE_TASK_EXECUTION_MAX_TASKS_DEFAULT);
  Preconditions.checkState(numExecutors >=1, "Must have at least 1 executor");
  ExecutorService rawExecutor = Executors.newFixedThreadPool(numExecutors,
      new ThreadFactoryBuilder().setDaemon(true).setNameFormat("LocalTaskExecutionThread #%d")
          .build());
  this.taskExecutorService = MoreExecutors.listeningDecorator(rawExecutor);
  boolean cleanupDagDataOnComplete = ShuffleUtils.isTezShuffleHandler(conf)
      && conf.getBoolean(TezConfiguration.TEZ_AM_DAG_CLEANUP_ON_COMPLETION,
      TezConfiguration.TEZ_AM_DAG_CLEANUP_ON_COMPLETION_DEFAULT);
  if (cleanupDagDataOnComplete) {
    String deletionTrackerClassName = conf.get(TezConfiguration.TEZ_AM_DELETION_TRACKER_CLASS,
        TezConfiguration.TEZ_AM_DELETION_TRACKER_CLASS_DEFAULT);
    deletionTracker = ReflectionUtils.createClazzInstance(
        deletionTrackerClassName, new Class[]{Configuration.class}, new Object[]{conf});
  }
}