Java 类org.apache.hadoop.mapreduce.v2.app.launcher.ContainerLauncher 实例源码

项目：hadoop 文件：TaskAttemptImpl.java

@SuppressWarnings("unchecked")
@Override
public void transition(TaskAttemptImpl taskAttempt, 
    TaskAttemptEvent event) {
  // unregister it to TaskAttemptListener so that it stops listening
  // for it
  taskAttempt.taskAttemptListener.unregister(
      taskAttempt.attemptId, taskAttempt.jvmID);

  if (event instanceof TaskAttemptKillEvent) {
    taskAttempt.addDiagnosticInfo(
        ((TaskAttemptKillEvent) event).getMessage());
  }

  taskAttempt.reportedStatus.progress = 1.0f;
  taskAttempt.updateProgressSplits();
  //send the cleanup event to containerLauncher
  taskAttempt.eventHandler.handle(new ContainerLauncherEvent(
      taskAttempt.attemptId, 
      taskAttempt.container.getId(), StringInterner
          .weakIntern(taskAttempt.container.getNodeId().toString()),
      taskAttempt.container.getContainerToken(),
      ContainerLauncher.EventType.CONTAINER_REMOTE_CLEANUP));
}

项目：hadoop 文件：TestRecovery.java

@Override
protected ContainerLauncher createContainerLauncher(AppContext context) {
  MockContainerLauncher launcher = new MockContainerLauncher() {
    @Override
    public void handle(ContainerLauncherEvent event) {
      TaskAttemptId taskAttemptID = event.getTaskAttemptID();
      // Pass everything except the 2nd attempt of the first task.
      if (taskAttemptID.getId() != 1
          || taskAttemptID.getTaskId().getId() != 0) {
        super.handle(event);
      }
    }
  };
  launcher.shufflePort = 5467;
  return launcher;
}

项目：hadoop 文件：TestMapReduceChildJVM.java

@Override
protected ContainerLauncher createContainerLauncher(AppContext context) {
  return new MockContainerLauncher() {
    @Override
    public void handle(ContainerLauncherEvent event) {
      if (event.getType() == EventType.CONTAINER_REMOTE_LAUNCH) {
        ContainerRemoteLaunchEvent launchEvent = (ContainerRemoteLaunchEvent) event;
        ContainerLaunchContext launchContext =
            launchEvent.getContainerLaunchContext();
        String cmdString = launchContext.getCommands().toString();
        LOG.info("launchContext " + cmdString);
        myCommandLine = cmdString;
        cmdEnvironment = launchContext.getEnvironment();
      }
      super.handle(event);
    }
  };
}

项目：aliyun-oss-hadoop-fs 文件：TaskAttemptImpl.java

@SuppressWarnings("unchecked")
@Override
public void transition(TaskAttemptImpl taskAttempt,
    TaskAttemptEvent event) {
  taskAttempt.setFinishTime();
  taskAttempt.taskAttemptListener.unregister(
      taskAttempt.attemptId, taskAttempt.jvmID);
  taskAttempt.eventHandler.handle(new ContainerLauncherEvent(
      taskAttempt.attemptId,
      taskAttempt.getAssignedContainerID(), taskAttempt.getAssignedContainerMgrAddress(),
      taskAttempt.container.getContainerToken(),
      ContainerLauncher.EventType.CONTAINER_REMOTE_CLEANUP));
  taskAttempt.eventHandler.handle(new TaskTAttemptEvent(
      taskAttempt.attemptId,
      TaskEventType.T_ATTEMPT_KILLED));

}

项目：aliyun-oss-hadoop-fs 文件：TestRecovery.java

@Override
protected ContainerLauncher createContainerLauncher(AppContext context) {
  MockContainerLauncher launcher = new MockContainerLauncher() {
    @Override
    public void handle(ContainerLauncherEvent event) {
      TaskAttemptId taskAttemptID = event.getTaskAttemptID();
      // Pass everything except the 2nd attempt of the first task.
      if (taskAttemptID.getId() != 1
          || taskAttemptID.getTaskId().getId() != 0) {
        super.handle(event);
      }
    }
  };
  launcher.shufflePort = 5467;
  return launcher;
}

项目：aliyun-oss-hadoop-fs 文件：TestMapReduceChildJVM.java

@Override
protected ContainerLauncher createContainerLauncher(AppContext context) {
  return new MockContainerLauncher() {
    @Override
    public void handle(ContainerLauncherEvent event) {
      if (event.getType() == EventType.CONTAINER_REMOTE_LAUNCH) {
        ContainerRemoteLaunchEvent launchEvent = (ContainerRemoteLaunchEvent) event;
        ContainerLaunchContext launchContext =
            launchEvent.getContainerLaunchContext();
        String cmdString = launchContext.getCommands().toString();
        LOG.info("launchContext " + cmdString);
        launchCmdList.add(cmdString);
        cmdEnvironment = launchContext.getEnvironment();
      }
      super.handle(event);
    }
  };
}

项目：big-c 文件：TaskAttemptImpl.java

@SuppressWarnings("unchecked")
@Override
public void transition(TaskAttemptImpl taskAttempt, 
    TaskAttemptEvent event) {
  // unregister it to TaskAttemptListener so that it stops listening
  // for it
  taskAttempt.taskAttemptListener.unregister(
      taskAttempt.attemptId, taskAttempt.jvmID);

  if (event instanceof TaskAttemptKillEvent) {
    taskAttempt.addDiagnosticInfo(
        ((TaskAttemptKillEvent) event).getMessage());
  }

  taskAttempt.reportedStatus.progress = 1.0f;
  taskAttempt.updateProgressSplits();
  //send the cleanup event to containerLauncher
  taskAttempt.eventHandler.handle(new ContainerLauncherEvent(
      taskAttempt.attemptId, 
      taskAttempt.container.getId(), StringInterner
          .weakIntern(taskAttempt.container.getNodeId().toString()),
      taskAttempt.container.getContainerToken(),
      ContainerLauncher.EventType.CONTAINER_REMOTE_CLEANUP));
}

项目：big-c 文件：TestRecovery.java

@Override
protected ContainerLauncher createContainerLauncher(AppContext context) {
  MockContainerLauncher launcher = new MockContainerLauncher() {
    @Override
    public void handle(ContainerLauncherEvent event) {
      TaskAttemptId taskAttemptID = event.getTaskAttemptID();
      // Pass everything except the 2nd attempt of the first task.
      if (taskAttemptID.getId() != 1
          || taskAttemptID.getTaskId().getId() != 0) {
        super.handle(event);
      }
    }
  };
  launcher.shufflePort = 5467;
  return launcher;
}

项目：big-c 文件：TestMapReduceChildJVM.java

@Override
protected ContainerLauncher createContainerLauncher(AppContext context) {
  return new MockContainerLauncher() {
    @Override
    public void handle(ContainerLauncherEvent event) {
      if (event.getType() == EventType.CONTAINER_REMOTE_LAUNCH) {
        ContainerRemoteLaunchEvent launchEvent = (ContainerRemoteLaunchEvent) event;
        ContainerLaunchContext launchContext =
            launchEvent.getContainerLaunchContext();
        String cmdString = launchContext.getCommands().toString();
        LOG.info("launchContext " + cmdString);
        myCommandLine = cmdString;
        cmdEnvironment = launchContext.getEnvironment();
      }
      super.handle(event);
    }
  };
}

项目：hadoop-2.6.0-cdh5.4.3 文件：TaskAttemptImpl.java

@SuppressWarnings("unchecked")
@Override
public void transition(TaskAttemptImpl taskAttempt, 
    TaskAttemptEvent event) {
  // unregister it to TaskAttemptListener so that it stops listening
  // for it
  taskAttempt.taskAttemptListener.unregister(
      taskAttempt.attemptId, taskAttempt.jvmID);

  if (event instanceof TaskAttemptKillEvent) {
    taskAttempt.addDiagnosticInfo(
        ((TaskAttemptKillEvent) event).getMessage());
  }

  taskAttempt.reportedStatus.progress = 1.0f;
  taskAttempt.updateProgressSplits();
  //send the cleanup event to containerLauncher
  taskAttempt.eventHandler.handle(new ContainerLauncherEvent(
      taskAttempt.attemptId, 
      taskAttempt.container.getId(), StringInterner
          .weakIntern(taskAttempt.container.getNodeId().toString()),
      taskAttempt.container.getContainerToken(),
      ContainerLauncher.EventType.CONTAINER_REMOTE_CLEANUP));
}

项目：hadoop-2.6.0-cdh5.4.3 文件：TestRecovery.java

@Override
protected ContainerLauncher createContainerLauncher(AppContext context) {
  MockContainerLauncher launcher = new MockContainerLauncher() {
    @Override
    public void handle(ContainerLauncherEvent event) {
      TaskAttemptId taskAttemptID = event.getTaskAttemptID();
      // Pass everything except the 2nd attempt of the first task.
      if (taskAttemptID.getId() != 1
          || taskAttemptID.getTaskId().getId() != 0) {
        super.handle(event);
      }
    }
  };
  launcher.shufflePort = 5467;
  return launcher;
}

项目：hadoop-2.6.0-cdh5.4.3 文件：TestMapReduceChildJVM.java

@Override
protected ContainerLauncher createContainerLauncher(AppContext context) {
  return new MockContainerLauncher() {
    @Override
    public void handle(ContainerLauncherEvent event) {
      if (event.getType() == EventType.CONTAINER_REMOTE_LAUNCH) {
        ContainerRemoteLaunchEvent launchEvent = (ContainerRemoteLaunchEvent) event;
        ContainerLaunchContext launchContext =
            launchEvent.getContainerLaunchContext();
        String cmdString = launchContext.getCommands().toString();
        LOG.info("launchContext " + cmdString);
        launchCmdList.add(cmdString);
        cmdEnvironment = launchContext.getEnvironment();
      }
      super.handle(event);
    }
  };
}

项目：hadoop-plus 文件：TaskAttemptImpl.java

@SuppressWarnings("unchecked")
@Override
public void transition(TaskAttemptImpl taskAttempt, 
    TaskAttemptEvent event) {
  // unregister it to TaskAttemptListener so that it stops listening
  // for it
  taskAttempt.taskAttemptListener.unregister(
      taskAttempt.attemptId, taskAttempt.jvmID);
  taskAttempt.reportedStatus.progress = 1.0f;
  taskAttempt.updateProgressSplits();
  //send the cleanup event to containerLauncher
  taskAttempt.eventHandler.handle(new ContainerLauncherEvent(
      taskAttempt.attemptId, 
      taskAttempt.container.getId(), StringInterner
          .weakIntern(taskAttempt.container.getNodeId().toString()),
      taskAttempt.container.getContainerToken(),
      ContainerLauncher.EventType.CONTAINER_REMOTE_CLEANUP));
}

项目：hadoop-plus 文件：TestRecovery.java

@Override
protected ContainerLauncher createContainerLauncher(AppContext context) {
  MockContainerLauncher launcher = new MockContainerLauncher() {
    @Override
    public void handle(ContainerLauncherEvent event) {
      TaskAttemptId taskAttemptID = event.getTaskAttemptID();
      // Pass everything except the 2nd attempt of the first task.
      if (taskAttemptID.getId() != 1
          || taskAttemptID.getTaskId().getId() != 0) {
        super.handle(event);
      }
    }
  };
  launcher.shufflePort = 5467;
  return launcher;
}

项目：hadoop-plus 文件：TestMapReduceChildJVM.java

@Override
protected ContainerLauncher createContainerLauncher(AppContext context) {
  return new MockContainerLauncher() {
    @Override
    public void handle(ContainerLauncherEvent event) {
      if (event.getType() == EventType.CONTAINER_REMOTE_LAUNCH) {
        ContainerRemoteLaunchEvent launchEvent = (ContainerRemoteLaunchEvent) event;
        ContainerLaunchContext launchContext =
            launchEvent.getContainerLaunchContext();
        String cmdString = launchContext.getCommands().toString();
        LOG.info("launchContext " + cmdString);
        myCommandLine = cmdString;
      }
      super.handle(event);
    }
  };
}

项目：FlexMap 文件：TestRecovery.java

@Override
protected ContainerLauncher createContainerLauncher(AppContext context) {
  MockContainerLauncher launcher = new MockContainerLauncher() {
    @Override
    public void handle(ContainerLauncherEvent event) {
      TaskAttemptId taskAttemptID = event.getTaskAttemptID();
      // Pass everything except the 2nd attempt of the first task.
      if (taskAttemptID.getId() != 1
          || taskAttemptID.getTaskId().getId() != 0) {
        super.handle(event);
      }
    }
  };
  launcher.shufflePort = 5467;
  return launcher;
}

项目：FlexMap 文件：TestMapReduceChildJVM.java

@Override
protected ContainerLauncher createContainerLauncher(AppContext context) {
  return new MockContainerLauncher() {
    @Override
    public void handle(ContainerLauncherEvent event) {
      if (event.getType() == EventType.CONTAINER_REMOTE_LAUNCH) {
        ContainerRemoteLaunchEvent launchEvent = (ContainerRemoteLaunchEvent) event;
        ContainerLaunchContext launchContext =
            launchEvent.getContainerLaunchContext();
        String cmdString = launchContext.getCommands().toString();
        LOG.info("launchContext " + cmdString);
        myCommandLine = cmdString;
        cmdEnvironment = launchContext.getEnvironment();
      }
      super.handle(event);
    }
  };
}

项目：hops 文件：TaskAttemptImpl.java

@SuppressWarnings("unchecked")
private static void sendContainerCleanup(TaskAttemptImpl taskAttempt,
    TaskAttemptEvent event) {
  if (event instanceof TaskAttemptKillEvent) {
    taskAttempt.addDiagnosticInfo(
        ((TaskAttemptKillEvent) event).getMessage());
  }
  //send the cleanup event to containerLauncher
  taskAttempt.eventHandler.handle(new ContainerLauncherEvent(
      taskAttempt.attemptId,
      taskAttempt.container.getId(), StringInterner
      .weakIntern(taskAttempt.container.getNodeId().toString()),
      taskAttempt.container.getContainerToken(),
      ContainerLauncher.EventType.CONTAINER_REMOTE_CLEANUP,
      event.getType() == TaskAttemptEventType.TA_TIMED_OUT));
}

项目：hops 文件：TestRecovery.java

@Override
protected ContainerLauncher createContainerLauncher(AppContext context) {
  MockContainerLauncher launcher = new MockContainerLauncher() {
    @Override
    public void handle(ContainerLauncherEvent event) {
      TaskAttemptId taskAttemptID = event.getTaskAttemptID();
      // Pass everything except the 2nd attempt of the first task.
      if (taskAttemptID.getId() != 1
          || taskAttemptID.getTaskId().getId() != 0) {
        super.handle(event);
      }
    }
  };
  launcher.shufflePort = 5467;
  return launcher;
}

项目：hops 文件：TestMapReduceChildJVM.java

@Override
protected ContainerLauncher createContainerLauncher(AppContext context) {
  return new MockContainerLauncher() {
    @Override
    public void handle(ContainerLauncherEvent event) {
      if (event.getType() == EventType.CONTAINER_REMOTE_LAUNCH) {
        ContainerRemoteLaunchEvent launchEvent = (ContainerRemoteLaunchEvent) event;
        ContainerLaunchContext launchContext =
            launchEvent.getContainerLaunchContext();
        String cmdString = launchContext.getCommands().toString();
        LOG.info("launchContext " + cmdString);
        myCommandLine = cmdString;
        cmdEnvironment = launchContext.getEnvironment();
      }
      super.handle(event);
    }
  };
}

项目：hadoop-TCP 文件：TaskAttemptImpl.java

@SuppressWarnings("unchecked")
@Override
public void transition(TaskAttemptImpl taskAttempt, 
    TaskAttemptEvent event) {
  // unregister it to TaskAttemptListener so that it stops listening
  // for it
  taskAttempt.taskAttemptListener.unregister(
      taskAttempt.attemptId, taskAttempt.jvmID);
  taskAttempt.reportedStatus.progress = 1.0f;
  taskAttempt.updateProgressSplits();
  //send the cleanup event to containerLauncher
  taskAttempt.eventHandler.handle(new ContainerLauncherEvent(
      taskAttempt.attemptId, 
      taskAttempt.container.getId(), StringInterner
          .weakIntern(taskAttempt.container.getNodeId().toString()),
      taskAttempt.container.getContainerToken(),
      ContainerLauncher.EventType.CONTAINER_REMOTE_CLEANUP));
}

项目：hadoop-TCP 文件：TestRecovery.java

@Override
protected ContainerLauncher createContainerLauncher(AppContext context) {
  MockContainerLauncher launcher = new MockContainerLauncher() {
    @Override
    public void handle(ContainerLauncherEvent event) {
      TaskAttemptId taskAttemptID = event.getTaskAttemptID();
      // Pass everything except the 2nd attempt of the first task.
      if (taskAttemptID.getId() != 1
          || taskAttemptID.getTaskId().getId() != 0) {
        super.handle(event);
      }
    }
  };
  launcher.shufflePort = 5467;
  return launcher;
}

项目：hadoop-TCP 文件：TestMapReduceChildJVM.java

@Override
protected ContainerLauncher createContainerLauncher(AppContext context) {
  return new MockContainerLauncher() {
    @Override
    public void handle(ContainerLauncherEvent event) {
      if (event.getType() == EventType.CONTAINER_REMOTE_LAUNCH) {
        ContainerRemoteLaunchEvent launchEvent = (ContainerRemoteLaunchEvent) event;
        ContainerLaunchContext launchContext =
            launchEvent.getContainerLaunchContext();
        String cmdString = launchContext.getCommands().toString();
        LOG.info("launchContext " + cmdString);
        myCommandLine = cmdString;
      }
      super.handle(event);
    }
  };
}

项目：hardfs 文件：TaskAttemptImpl.java

@SuppressWarnings("unchecked")
@Override
public void transition(TaskAttemptImpl taskAttempt, 
    TaskAttemptEvent event) {
  // unregister it to TaskAttemptListener so that it stops listening
  // for it
  taskAttempt.taskAttemptListener.unregister(
      taskAttempt.attemptId, taskAttempt.jvmID);
  taskAttempt.reportedStatus.progress = 1.0f;
  taskAttempt.updateProgressSplits();
  //send the cleanup event to containerLauncher
  taskAttempt.eventHandler.handle(new ContainerLauncherEvent(
      taskAttempt.attemptId, 
      taskAttempt.container.getId(), StringInterner
          .weakIntern(taskAttempt.container.getNodeId().toString()),
      taskAttempt.container.getContainerToken(),
      ContainerLauncher.EventType.CONTAINER_REMOTE_CLEANUP));
}

项目：hardfs 文件：TestRecovery.java

@Override
protected ContainerLauncher createContainerLauncher(AppContext context) {
  MockContainerLauncher launcher = new MockContainerLauncher() {
    @Override
    public void handle(ContainerLauncherEvent event) {
      TaskAttemptId taskAttemptID = event.getTaskAttemptID();
      // Pass everything except the 2nd attempt of the first task.
      if (taskAttemptID.getId() != 1
          || taskAttemptID.getTaskId().getId() != 0) {
        super.handle(event);
      }
    }
  };
  launcher.shufflePort = 5467;
  return launcher;
}

项目：hardfs 文件：TestMapReduceChildJVM.java

@Override
protected ContainerLauncher createContainerLauncher(AppContext context) {
  return new MockContainerLauncher() {
    @Override
    public void handle(ContainerLauncherEvent event) {
      if (event.getType() == EventType.CONTAINER_REMOTE_LAUNCH) {
        ContainerRemoteLaunchEvent launchEvent = (ContainerRemoteLaunchEvent) event;
        ContainerLaunchContext launchContext =
            launchEvent.getContainerLaunchContext();
        String cmdString = launchContext.getCommands().toString();
        LOG.info("launchContext " + cmdString);
        myCommandLine = cmdString;
      }
      super.handle(event);
    }
  };
}

项目：hadoop-on-lustre2 文件：TaskAttemptImpl.java

@SuppressWarnings("unchecked")
@Override
public void transition(TaskAttemptImpl taskAttempt, 
    TaskAttemptEvent event) {
  // unregister it to TaskAttemptListener so that it stops listening
  // for it
  taskAttempt.taskAttemptListener.unregister(
      taskAttempt.attemptId, taskAttempt.jvmID);

  if (event instanceof TaskAttemptKillEvent) {
    taskAttempt.addDiagnosticInfo(
        ((TaskAttemptKillEvent) event).getMessage());
  }

  taskAttempt.reportedStatus.progress = 1.0f;
  taskAttempt.updateProgressSplits();
  //send the cleanup event to containerLauncher
  taskAttempt.eventHandler.handle(new ContainerLauncherEvent(
      taskAttempt.attemptId, 
      taskAttempt.container.getId(), StringInterner
          .weakIntern(taskAttempt.container.getNodeId().toString()),
      taskAttempt.container.getContainerToken(),
      ContainerLauncher.EventType.CONTAINER_REMOTE_CLEANUP));
}

项目：hadoop-on-lustre2 文件：TestRecovery.java

@Override
protected ContainerLauncher createContainerLauncher(AppContext context) {
  MockContainerLauncher launcher = new MockContainerLauncher() {
    @Override
    public void handle(ContainerLauncherEvent event) {
      TaskAttemptId taskAttemptID = event.getTaskAttemptID();
      // Pass everything except the 2nd attempt of the first task.
      if (taskAttemptID.getId() != 1
          || taskAttemptID.getTaskId().getId() != 0) {
        super.handle(event);
      }
    }
  };
  launcher.shufflePort = 5467;
  return launcher;
}

项目：hadoop-on-lustre2 文件：TestMapReduceChildJVM.java

@Override
protected ContainerLauncher createContainerLauncher(AppContext context) {
  return new MockContainerLauncher() {
    @Override
    public void handle(ContainerLauncherEvent event) {
      if (event.getType() == EventType.CONTAINER_REMOTE_LAUNCH) {
        ContainerRemoteLaunchEvent launchEvent = (ContainerRemoteLaunchEvent) event;
        ContainerLaunchContext launchContext =
            launchEvent.getContainerLaunchContext();
        String cmdString = launchContext.getCommands().toString();
        LOG.info("launchContext " + cmdString);
        myCommandLine = cmdString;
        cmdEnvironment = launchContext.getEnvironment();
      }
      super.handle(event);
    }
  };
}

项目：hadoop 文件：TestMRApp.java

@Test
public void testContainerPassThrough() throws Exception {
  MRApp app = new MRApp(0, 1, true, this.getClass().getName(), true) {
    @Override
    protected ContainerLauncher createContainerLauncher(AppContext context) {
      return new MockContainerLauncher() {
        @Override
        public void handle(ContainerLauncherEvent event) {
          if (event instanceof ContainerRemoteLaunchEvent) {
            containerObtainedByContainerLauncher =
                ((ContainerRemoteLaunchEvent) event).getAllocatedContainer();
          }
          super.handle(event);
        }
      };
    };
  };
  Job job = app.submit(new Configuration());
  app.waitForState(job, JobState.SUCCEEDED);
  app.verifyCompleted();

  Collection<Task> tasks = job.getTasks().values();
  Collection<TaskAttempt> taskAttempts =
      tasks.iterator().next().getAttempts().values();
  TaskAttemptImpl taskAttempt =
      (TaskAttemptImpl) taskAttempts.iterator().next();
  // Container from RM should pass through to the launcher. Container object
  // should be the same.
 Assert.assertTrue(taskAttempt.container 
   == containerObtainedByContainerLauncher);
}

项目：hadoop 文件：TestFail.java

@Override
protected ContainerLauncher createContainerLauncher(AppContext context) {
  return new ContainerLauncherImpl(context) {
    @Override
    public void handle(ContainerLauncherEvent event) {

      switch (event.getType()) {
      case CONTAINER_REMOTE_LAUNCH:
        super.handle(event); // Unused event and container.
        break;
      case CONTAINER_REMOTE_CLEANUP:
        getContext().getEventHandler().handle(
            new TaskAttemptEvent(event.getTaskAttemptID(),
                TaskAttemptEventType.TA_CONTAINER_CLEANED));
        break;
      }
    }

    @Override
    public ContainerManagementProtocolProxyData getCMProxy(
        String containerMgrBindAddr, ContainerId containerId)
        throws IOException {
      try {
        synchronized (this) {
          wait(); // Just hang the thread simulating a very slow NM.
        }
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
      return null;
    }
  };
}

项目：aliyun-oss-hadoop-fs 文件：TaskAttemptImpl.java

@SuppressWarnings("unchecked")
private static void sendContainerCompleted(TaskAttemptImpl taskAttempt) {
  taskAttempt.eventHandler.handle(new ContainerLauncherEvent(
      taskAttempt.attemptId,
      taskAttempt.container.getId(), StringInterner
      .weakIntern(taskAttempt.container.getNodeId().toString()),
      taskAttempt.container.getContainerToken(),
      ContainerLauncher.EventType.CONTAINER_COMPLETED));
}

项目：aliyun-oss-hadoop-fs 文件：TaskAttemptImpl.java

@SuppressWarnings("unchecked")
private static void sendContainerCleanup(TaskAttemptImpl taskAttempt,
    TaskAttemptEvent event) {
  if (event instanceof TaskAttemptKillEvent) {
    taskAttempt.addDiagnosticInfo(
        ((TaskAttemptKillEvent) event).getMessage());
  }
  //send the cleanup event to containerLauncher
  taskAttempt.eventHandler.handle(new ContainerLauncherEvent(
      taskAttempt.attemptId,
      taskAttempt.container.getId(), StringInterner
      .weakIntern(taskAttempt.container.getNodeId().toString()),
      taskAttempt.container.getContainerToken(),
      ContainerLauncher.EventType.CONTAINER_REMOTE_CLEANUP));
}

项目：aliyun-oss-hadoop-fs 文件：TestMRApp.java

@Test
public void testContainerPassThrough() throws Exception {
  MRApp app = new MRApp(0, 1, true, this.getClass().getName(), true) {
    @Override
    protected ContainerLauncher createContainerLauncher(AppContext context) {
      return new MockContainerLauncher() {
        @Override
        public void handle(ContainerLauncherEvent event) {
          if (event instanceof ContainerRemoteLaunchEvent) {
            containerObtainedByContainerLauncher =
                ((ContainerRemoteLaunchEvent) event).getAllocatedContainer();
          }
          super.handle(event);
        }
      };
    };
  };
  Job job = app.submit(new Configuration());
  app.waitForState(job, JobState.SUCCEEDED);
  app.verifyCompleted();

  Collection<Task> tasks = job.getTasks().values();
  Collection<TaskAttempt> taskAttempts =
      tasks.iterator().next().getAttempts().values();
  TaskAttemptImpl taskAttempt =
      (TaskAttemptImpl) taskAttempts.iterator().next();
  // Container from RM should pass through to the launcher. Container object
  // should be the same.
 Assert.assertTrue(taskAttempt.container 
   == containerObtainedByContainerLauncher);
}

项目：aliyun-oss-hadoop-fs 文件：TestFail.java

@Override
protected ContainerLauncher createContainerLauncher(AppContext context) {
  return new ContainerLauncherImpl(context) {
    @Override
    public void handle(ContainerLauncherEvent event) {

      switch (event.getType()) {
      case CONTAINER_REMOTE_LAUNCH:
        super.handle(event); // Unused event and container.
        break;
      case CONTAINER_REMOTE_CLEANUP:
        getContext().getEventHandler().handle(
            new TaskAttemptEvent(event.getTaskAttemptID(),
                TaskAttemptEventType.TA_CONTAINER_CLEANED));
        break;
      case CONTAINER_COMPLETED:
        super.handle(event);
      }
    }

    @Override
    public ContainerManagementProtocolProxyData getCMProxy(
        String containerMgrBindAddr, ContainerId containerId)
        throws IOException {
      try {
        synchronized (this) {
          wait(); // Just hang the thread simulating a very slow NM.
        }
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
      return null;
    }
  };
}

项目：big-c 文件：TestMRApp.java

@Test
public void testContainerPassThrough() throws Exception {
  MRApp app = new MRApp(0, 1, true, this.getClass().getName(), true) {
    @Override
    protected ContainerLauncher createContainerLauncher(AppContext context) {
      return new MockContainerLauncher() {
        @Override
        public void handle(ContainerLauncherEvent event) {
          if (event instanceof ContainerRemoteLaunchEvent) {
            containerObtainedByContainerLauncher =
                ((ContainerRemoteLaunchEvent) event).getAllocatedContainer();
          }
          super.handle(event);
        }
      };
    };
  };
  Job job = app.submit(new Configuration());
  app.waitForState(job, JobState.SUCCEEDED);
  app.verifyCompleted();

  Collection<Task> tasks = job.getTasks().values();
  Collection<TaskAttempt> taskAttempts =
      tasks.iterator().next().getAttempts().values();
  TaskAttemptImpl taskAttempt =
      (TaskAttemptImpl) taskAttempts.iterator().next();
  // Container from RM should pass through to the launcher. Container object
  // should be the same.
 Assert.assertTrue(taskAttempt.container 
   == containerObtainedByContainerLauncher);
}

项目：big-c 文件：TestFail.java

@Override
protected ContainerLauncher createContainerLauncher(AppContext context) {
  return new ContainerLauncherImpl(context) {
    @Override
    public void handle(ContainerLauncherEvent event) {

      switch (event.getType()) {
      case CONTAINER_REMOTE_LAUNCH:
        super.handle(event); // Unused event and container.
        break;
      case CONTAINER_REMOTE_CLEANUP:
        getContext().getEventHandler().handle(
            new TaskAttemptEvent(event.getTaskAttemptID(),
                TaskAttemptEventType.TA_CONTAINER_CLEANED));
        break;
      }
    }

    @Override
    public ContainerManagementProtocolProxyData getCMProxy(
        String containerMgrBindAddr, ContainerId containerId)
        throws IOException {
      try {
        synchronized (this) {
          wait(); // Just hang the thread simulating a very slow NM.
        }
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
      return null;
    }
  };
}

项目：hadoop-2.6.0-cdh5.4.3 文件：TestMRApp.java

@Test
public void testContainerPassThrough() throws Exception {
  MRApp app = new MRApp(0, 1, true, this.getClass().getName(), true) {
    @Override
    protected ContainerLauncher createContainerLauncher(AppContext context) {
      return new MockContainerLauncher() {
        @Override
        public void handle(ContainerLauncherEvent event) {
          if (event instanceof ContainerRemoteLaunchEvent) {
            containerObtainedByContainerLauncher =
                ((ContainerRemoteLaunchEvent) event).getAllocatedContainer();
          }
          super.handle(event);
        }
      };
    };
  };
  Job job = app.submit(new Configuration());
  app.waitForState(job, JobState.SUCCEEDED);
  app.verifyCompleted();

  Collection<Task> tasks = job.getTasks().values();
  Collection<TaskAttempt> taskAttempts =
      tasks.iterator().next().getAttempts().values();
  TaskAttemptImpl taskAttempt =
      (TaskAttemptImpl) taskAttempts.iterator().next();
  // Container from RM should pass through to the launcher. Container object
  // should be the same.
 Assert.assertTrue(taskAttempt.container 
   == containerObtainedByContainerLauncher);
}

项目：hadoop-2.6.0-cdh5.4.3 文件：TestFail.java

@Override
protected ContainerLauncher createContainerLauncher(AppContext context) {
  return new ContainerLauncherImpl(context) {
    @Override
    public void handle(ContainerLauncherEvent event) {

      switch (event.getType()) {
      case CONTAINER_REMOTE_LAUNCH:
        super.handle(event); // Unused event and container.
        break;
      case CONTAINER_REMOTE_CLEANUP:
        getContext().getEventHandler().handle(
            new TaskAttemptEvent(event.getTaskAttemptID(),
                TaskAttemptEventType.TA_CONTAINER_CLEANED));
        break;
      }
    }

    @Override
    public ContainerManagementProtocolProxyData getCMProxy(
        String containerMgrBindAddr, ContainerId containerId)
        throws IOException {
      try {
        synchronized (this) {
          wait(); // Just hang the thread simulating a very slow NM.
        }
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
      return null;
    }
  };
}

项目：hadoop-plus 文件：TestMRApp.java

@Test
public void testContainerPassThrough() throws Exception {
  MRApp app = new MRApp(0, 1, true, this.getClass().getName(), true) {
    @Override
    protected ContainerLauncher createContainerLauncher(AppContext context) {
      return new MockContainerLauncher() {
        @Override
        public void handle(ContainerLauncherEvent event) {
          if (event instanceof ContainerRemoteLaunchEvent) {
            containerObtainedByContainerLauncher =
                ((ContainerRemoteLaunchEvent) event).getAllocatedContainer();
          }
          super.handle(event);
        }
      };
    };
  };
  Job job = app.submit(new Configuration());
  app.waitForState(job, JobState.SUCCEEDED);
  app.verifyCompleted();

  Collection<Task> tasks = job.getTasks().values();
  Collection<TaskAttempt> taskAttempts =
      tasks.iterator().next().getAttempts().values();
  TaskAttemptImpl taskAttempt =
      (TaskAttemptImpl) taskAttempts.iterator().next();
  // Container from RM should pass through to the launcher. Container object
  // should be the same.
 Assert.assertTrue(taskAttempt.container 
   == containerObtainedByContainerLauncher);
}