import org.apache.twill.api.TwillController; //导入依赖的package包/类
private void stopClusterAsync(Cluster cluster) throws YarnProvisioningHandlingException {
  TwillController controller = getTwillControllerHelper(cluster);
  if (cluster.getState() == ClusterState.STOPPED || cluster.getState() == ClusterState.FAILED) {
    // nothing to do - probably termination routine was called already
    return;
  }
  if (controller == null) {
    if (cluster.getState() != ClusterState.STOPPED && cluster.getState() != ClusterState.FAILED) {
      logger.warn("Either cluster is already stopped or YarnTwillRunnerService was not initialized yet. You may want to try again");
      cluster.setState(ClusterState.STOPPED);
    }
    return;
  }
  // async call
  cluster.setState(ClusterState.STOPPING);
  controller.terminate();
  return;
}
 

import org.apache.twill.api.TwillController; //导入依赖的package包/类
private TwillController getTwillControllerHelper(Cluster cluster) throws YarnProvisioningHandlingException {
  RunId runId = cluster.getRunId();
  if (runId == null) {
    return null;
  }

  TwillRunnerService twillService = getTwillService(cluster);
  if (twillService == null) {
    logger.error("YarnTwillRunnerService is null. Possibly was not instantiated yet");
    return null;
  }

  String applicationName = cluster.getClusterConfig().getName();
  if (applicationName == null) {
    applicationName = DacDaemonYarnApplication.YARN_APPLICATION_NAME_DEFAULT;
  }

  TwillController controller =  twillService.lookup(applicationName, RunIds.fromString(runId.getId()));
  initOnTerminatingThread(cluster,controller);

  return controller;
}
 

import org.apache.twill.api.TwillController; //导入依赖的package包/类
@Test
public void testStopCluster() throws Exception {
  assumeNonMaprProfile();

  Cluster myCluster = createCluster();
  myCluster.setState(ClusterState.RUNNING);

  YarnController controller = Mockito.mock(YarnController.class);
  YarnService yarnService = new YarnService(new TestListener(), controller, Mockito.mock(NodeProvider.class));

  TwillController twillController = Mockito.mock(TwillController.class);
  RunId runId = RunIds.generate();

  when(controller.startCluster(any(YarnConfiguration.class), eq(myCluster.getClusterConfig().getSubPropertyList())))
    .thenReturn(twillController);
  when(twillController.getRunId()).thenReturn(runId);

  myCluster.setRunId(new com.dremio.provision.RunId(runId.getId()));
  yarnService.stopCluster(myCluster);

  assertEquals(ClusterState.STOPPED, myCluster.getState());
}
 

import org.apache.twill.api.TwillController; //导入依赖的package包/类
@Ignore
@Test
public void testDistributedShell() throws InterruptedException {
  TwillRunner twillRunner = getTwillRunner();

  TwillController controller = twillRunner.prepare(new DistributedShell("pwd", "ls -al"))
                                          .addLogHandler(new PrinterLogHandler(new PrintWriter(System.out)))
                                          .start();

  final CountDownLatch stopLatch = new CountDownLatch(1);
  controller.onTerminated(new Runnable() {
    @Override
    public void run() {
      stopLatch.countDown();
    }
  }, Threads.SAME_THREAD_EXECUTOR);

  Assert.assertTrue(stopLatch.await(10, TimeUnit.SECONDS));
}
 

import org.apache.twill.api.TwillController; //导入依赖的package包/类
private boolean waitForDebugPort(TwillController controller, String runnable, int timeLimit)
  throws InterruptedException {
  long millis = 0;
  while (millis < 1000 * timeLimit) {
    ResourceReport report = controller.getResourceReport();
    if (report == null || report.getRunnableResources(runnable) == null) {
      continue;
    }
    for (TwillRunResources resources : report.getRunnableResources(runnable)) {
      if (resources.getDebugPort() != null) {
        return true;
      }
    }
    TimeUnit.MILLISECONDS.sleep(100);
    millis += 100;
  }
  return false;
}
 

import org.apache.twill.api.TwillController; //导入依赖的package包/类
@Test
public void testDebugPortOneRunnable() throws Exception {
  YarnTwillRunnerService runner = getTwillRunner();
  runner.start();

  TwillController controller = runner.prepare(new DummyApplication())
                                     .enableDebugging("r1")
                                     .addLogHandler(new PrinterLogHandler(new PrintWriter(System.out)))
                                     .start();
  final CountDownLatch running = new CountDownLatch(1);
  controller.onRunning(new Runnable() {
    @Override
    public void run() {
      running.countDown();
    }
  }, Threads.SAME_THREAD_EXECUTOR);

  Assert.assertTrue(running.await(120, TimeUnit.SECONDS));
  Assert.assertTrue(waitForDebugPort(controller, "r1", 30));
  controller.terminate().get(120, TimeUnit.SECONDS);
  // Sleep a bit before exiting.
  TimeUnit.SECONDS.sleep(2);
}
 

import org.apache.twill.api.TwillController; //导入依赖的package包/类
@Test
public void testDebugPortAllRunnables() throws Exception {
  YarnTwillRunnerService runner = getTwillRunner();
  runner.start();

  TwillController controller = runner.prepare(new DummyApplication())
                                     .enableDebugging()
                                     .addLogHandler(new PrinterLogHandler(new PrintWriter(System.out)))
                                     .start();
  final CountDownLatch running = new CountDownLatch(1);
  controller.onRunning(new Runnable() {
    @Override
    public void run() {
      running.countDown();
    }
  }, Threads.SAME_THREAD_EXECUTOR);

  Assert.assertTrue(running.await(120, TimeUnit.SECONDS));
  Assert.assertTrue(waitForDebugPort(controller, "r1", 30));
  Assert.assertTrue(waitForDebugPort(controller, "r2", 30));
  controller.terminate().get(120, TimeUnit.SECONDS);
  // Sleep a bit before exiting.
  TimeUnit.SECONDS.sleep(2);
}
 

import org.apache.twill.api.TwillController; //导入依赖的package包/类
@Test
public void testTaskCompleted() throws InterruptedException, TimeoutException, ExecutionException {
  TwillRunner twillRunner = getTwillRunner();
  TwillController controller = twillRunner.prepare(new SleepTask(),
                                              ResourceSpecification.Builder.with()
                                                .setVirtualCores(1)
                                                .setMemory(512, ResourceSpecification.SizeUnit.MEGA)
                                                .setInstances(3).build())
                                          .addLogHandler(new PrinterLogHandler(new PrintWriter(System.out, true)))
                                          .start();

  final CountDownLatch runLatch = new CountDownLatch(1);
  controller.onRunning(new Runnable() {
    @Override
    public void run() {
      runLatch.countDown();
    }
  }, Threads.SAME_THREAD_EXECUTOR);

  Assert.assertTrue(runLatch.await(1, TimeUnit.MINUTES));
  controller.awaitTerminated(1, TimeUnit.MINUTES);
  Assert.assertEquals(ServiceController.TerminationStatus.SUCCEEDED, controller.getTerminationStatus());
}
 

import org.apache.twill.api.TwillController; //导入依赖的package包/类
@Test
public void testFailureComplete() throws TimeoutException, ExecutionException, InterruptedException {
  TwillRunner twillRunner = getTwillRunner();

  // Start the app with an invalid ClassLoader. This will cause the AM fails to start.
  TwillController controller = twillRunner.prepare(new SleepTask(),
                                                   ResourceSpecification.Builder.with()
                                                     .setVirtualCores(1)
                                                     .setMemory(512, ResourceSpecification.SizeUnit.MEGA)
                                                     .setInstances(1).build())
    .addLogHandler(new PrinterLogHandler(new PrintWriter(System.out, true)))
    .setClassLoader("InvalidClassLoader")
    .start();

  final CountDownLatch terminateLatch = new CountDownLatch(1);
  controller.onTerminated(new Runnable() {
    @Override
    public void run() {
      terminateLatch.countDown();
    }
  }, Threads.SAME_THREAD_EXECUTOR);

  Assert.assertTrue(terminateLatch.await(2, TimeUnit.MINUTES));
  Assert.assertEquals(ServiceController.TerminationStatus.FAILED, controller.getTerminationStatus());
}
 

import org.apache.twill.api.TwillController; //导入依赖的package包/类
@Test
public void testInitFail() throws InterruptedException, ExecutionException, TimeoutException {
  TwillRunner runner = getTwillRunner();
  final CountDownLatch logLatch = new CountDownLatch(1);

  // Verify that it receives the exception log entry that thrown when runnable initialize
  LogHandler logVerifyHandler = new LogHandler() {
    @Override
    public void onLog(LogEntry logEntry) {
      if (logEntry.getMessage().endsWith("exited abnormally with state COMPLETE, exit code 10.")) {
        logLatch.countDown();
      }
    }
  };

  TwillController controller = runner
    .prepare(new InitFailRunnable())
    .addLogHandler(new PrinterLogHandler(new PrintWriter(System.out)))
    .addLogHandler(logVerifyHandler)
    .start();

  controller.awaitTerminated(2, TimeUnit.MINUTES);
  Assert.assertTrue(logLatch.await(10, TimeUnit.SECONDS));
}
 

import org.apache.twill.api.TwillController; //导入依赖的package包/类
@Test
public void testKilled() throws IOException, InterruptedException, TimeoutException, ExecutionException {
  // Create a parent folder to be written by EventHandler
  File parentFolder = TMP_FOLDER.newFolder();
  parentFolder.setWritable(true, false);
  TwillController controller = getTwillRunner().prepare(new SleepApplication(parentFolder.getAbsolutePath()))
    .addLogHandler(new PrinterLogHandler(new PrintWriter(System.out, true)))
    .start();
  // Wait for the runnable to run and create runFile within 120 secs
  File runFile = new File(parentFolder, RUN_FILE);
  Stopwatch stopwatch = new Stopwatch().start();
  while (!runFile.exists() && stopwatch.elapsedTime(TimeUnit.SECONDS) < 120) {
    TimeUnit.SECONDS.sleep(1);
  }
  Assert.assertTrue(runFile.exists());
  // Terminate the app once the runnable runs
  controller.terminate();
  controller.awaitTerminated(120, TimeUnit.SECONDS);
  // EventHandler#killed() method should be called to create a file
  Assert.assertTrue(new File(parentFolder.getAbsolutePath(), KILLED_FILE).exists());
  // EventHandler#completed() method should not be called
  Assert.assertFalse(new File(parentFolder.getAbsolutePath(), COMPLETED_FILE).exists());
  // EventHandler#aborted() method should not be called
  Assert.assertFalse(new File(parentFolder.getAbsolutePath(), ABORTED_FILE).exists());
}
 

import org.apache.twill.api.TwillController; //导入依赖的package包/类
@Test
public void testExtraOptions() throws InterruptedException, ExecutionException {
  // Start the testing app with jvm options at both global level as well as for the specific runnables.
  TwillController controller = getTwillRunner()
    .prepare(new JvmOptionsApplication())
    .addLogHandler(new PrinterLogHandler(new PrintWriter(System.out)))
    .setJVMOptions("-Dservice.name=default")
    .setJVMOptions("r2", "-Dservice.name=r2")
    .start();

  // For r1 and r3 will be using "default" as the service name.
  waitForSize(controller.discoverService("default"), 2, 120);
  // r2 will be use "r2" as the service name.
  waitForSize(controller.discoverService("r2"), 1, 120);

  controller.terminate().get();
}
 

import org.apache.twill.api.TwillController; //导入依赖的package包/类
private boolean waitForLogLevel(TwillController controller, String runnable, long timeout,
                                TimeUnit timeoutUnit, @Nullable LogEntry.Level expected) throws InterruptedException {

  Stopwatch stopwatch = new Stopwatch();
  stopwatch.start();
  do {
    ResourceReport report = controller.getResourceReport();
    if (report == null || report.getRunnableResources(runnable) == null) {
      continue;
    }
    for (TwillRunResources resources : report.getRunnableResources(runnable)) {
      LogEntry.Level level = resources.getLogLevels().get(Logger.ROOT_LOGGER_NAME);
      if (expected == level) {
         return true;
      }
    }
    TimeUnit.MILLISECONDS.sleep(100);
  } while (stopwatch.elapsedTime(timeoutUnit) < timeout);

  return false;
}
 

import org.apache.twill.api.TwillController; //导入依赖的package包/类
private void waitForContainers(TwillController controller, int count, long timeout, TimeUnit timeoutUnit)
  throws Exception {
  Stopwatch stopwatch = new Stopwatch();
  stopwatch.start();
  int yarnContainers = 0;
  int twillContainers = 0;
  do {
    if (controller.getResourceReport() != null) {
      yarnContainers =
        getApplicationResourceReport(controller.getResourceReport().getApplicationId()).getNumUsedContainers();
      twillContainers = getTwillContainersUsed(controller);
      if (yarnContainers == count && twillContainers == count) {
        return;
      }
    }
    TimeUnit.SECONDS.sleep(1);
  } while (stopwatch.elapsedTime(timeoutUnit) < timeout);

  throw new TimeoutException("Timeout reached while waiting for num containers to be " +  count +
                               ". Yarn containers = " + yarnContainers + ", Twill containers = " + twillContainers);
}
 

import org.apache.twill.api.TwillController; //导入依赖的package包/类
private void waitForInstance(TwillController controller, String runnable, String yarnInstanceId,
                             long timeout, TimeUnit timeoutUnit) throws InterruptedException, TimeoutException {
  Stopwatch stopwatch = new Stopwatch();
  stopwatch.start();
  do {
    ResourceReport report = controller.getResourceReport();
    if (report != null && report.getRunnableResources(runnable) != null) {
      for (TwillRunResources resources : report.getRunnableResources(runnable)) {
        if (resources.getContainerId().endsWith(yarnInstanceId)) {
          return;
        }
      }
    }
    TimeUnit.SECONDS.sleep(1);
  } while (stopwatch.elapsedTime(timeoutUnit) < timeout);

  throw new TimeoutException("Timeout reached while waiting for runnable " +
                               runnable + " instance " + yarnInstanceId);
}
 

import org.apache.twill.api.TwillController; //导入依赖的package包/类
@Test
public void testProvisionTimeout() throws InterruptedException, ExecutionException, TimeoutException, IOException {
  TwillRunner runner = getTwillRunner();
  // Create a parent folder to be written by EventHandler#aborted()
  File parentFolder = TMP_FOLDER.newFolder();
  parentFolder.setWritable(true, false);
  TwillController controller = runner.prepare(new TimeoutApplication(parentFolder.getAbsolutePath()))
                                     .addLogHandler(new PrinterLogHandler(new PrintWriter(System.out, true)))
                                     .start();

  // The provision should failed in 30 seconds after AM started, which AM could took a while to start.
  // Hence we give 90 seconds max time here.
  try {
    controller.awaitTerminated(90, TimeUnit.SECONDS);
    // EventHandler#aborted() method should be called to create a file
    Assert.assertTrue(new File(parentFolder.getAbsolutePath(), ABORTED_FILE).exists());
    String[] abortedFiles = parentFolder.list();
    Assert.assertNotNull(abortedFiles);
    Assert.assertEquals(1, abortedFiles.length);
  } finally {
    // If it timeout, kill the app as cleanup.
    controller.kill();
  }
}
 

import org.apache.twill.api.TwillController; //导入依赖的package包/类
private void maxRetriesRun(final int instances) throws TimeoutException, ExecutionException {
  TwillRunner runner = getTwillRunner();
  final int maxRetries = 3;
  final AtomicInteger retriesSeen = new AtomicInteger(0);

  ResourceSpecification resource = ResourceSpecification.Builder.with().setVirtualCores(1)
    .setMemory(512, ResourceSpecification.SizeUnit.MEGA).setInstances(instances).build();

  TwillController controller = runner.prepare(new FailingServer(), resource)
    .withMaxRetries(FailingServer.class.getSimpleName(), maxRetries)
    .addLogHandler(new PrinterLogHandler(new PrintWriter(System.out, true)))
    .addLogHandler(new LogHandler() {
      @Override
      public void onLog(LogEntry logEntry) {
        if (logEntry.getMessage().contains("retries for instance")) {
          retriesSeen.incrementAndGet();
        }
      }
    })
    .start();

  controller.awaitTerminated(2, TimeUnit.MINUTES);
  Assert.assertEquals(maxRetries * instances, retriesSeen.get());
}
 

import org.apache.twill.api.TwillController; //导入依赖的package包/类
@Test
public void test() throws ExecutionException, InterruptedException, TimeoutException {
  TwillRunner runner = getTwillRunner();

  // Start the TestRunnable and make sure it is executed with the log message emitted.
  CountDownLatch logLatch = new CountDownLatch(1);
  TwillController controller = runner.prepare(new TestRunnable())
    .addLogHandler(new PrinterLogHandler(new PrintWriter(System.out, true)))
    .addLogHandler(logEntry -> {
      if ("Hello World".equals(logEntry.getMessage())) {
        logLatch.countDown();
      }
    })
    .start();

  Assert.assertTrue(logLatch.await(120, TimeUnit.SECONDS));
  controller.terminate().get(120, TimeUnit.SECONDS);
}
 

import org.apache.twill.api.TwillController; //导入依赖的package包/类
/**
 * Terminates all instances of the {@link PeriodicNotificationTwillApp} on the YARN cluster.
 */
public void stopApp() {
    LOG.info("Stopping any running instances...");

    int counter = 0;
    // It is possible that we have launched multiple instances of the app.  For now, stop them all, one at a time.
    for(final TwillController c : twillRunner.lookup(PeriodicNotificationTwillApp.APPLICATION_NAME)) {
        final ResourceReport report = c.getResourceReport();
        LOG.info("Attempting to stop {} with YARN ApplicationId: {} and Twill RunId: {}", PeriodicNotificationTwillApp.APPLICATION_NAME, report.getApplicationId(), c.getRunId());
        Futures.getUnchecked(c.terminate());
        LOG.info("Stopped {} with YARN ApplicationId: {} and Twill RunId: {}", PeriodicNotificationTwillApp.APPLICATION_NAME, report.getApplicationId(), c.getRunId());
        counter++;
    }

    LOG.info("Stopped {} instance(s) of {}", counter, PeriodicNotificationTwillApp.APPLICATION_NAME);
}
 

import org.apache.twill.api.TwillController; //导入依赖的package包/类
/**
 * Blocks until a non-null Resource report is returned.
 * @param controller - The controller to interrogate.
 * @param timeout - The maximum time to poll {@controller}.  Use -1 for infinite polling.
 * @param timeoutUnits - The units of {@code timeout}.
 * @return The ResourceReport for the application.
 * @throws IllegalStateException If a timeout occurs before a ResourceReport is returned.
 */
private ResourceReport getResourceReport(final TwillController controller, final long timeout, final TimeUnit timeoutUnits) {
    Preconditions.checkArgument(timeout >= -1, "timeout cannot be less than -1");
    final long timeoutMillis = TimeUnit.MILLISECONDS.convert(timeout, timeoutUnits);
    final long sleepMillis = 1000; // how long to sleep between retrieval attempts.
    long totalElapsedMillis = 0;
    ResourceReport report = controller.getResourceReport();
    while (reportIsLoading(report)) {
        try {
            Thread.sleep(sleepMillis);
        } catch (final InterruptedException e) {
            throw new IllegalStateException(e);
        }
        totalElapsedMillis += sleepMillis;
        if ((timeout != -1) && (totalElapsedMillis >= timeoutMillis)) {
            final String errorMessage = "Timeout while waiting for the Twill Application to start on YARN.  Total elapsed time: " + TimeUnit.SECONDS.convert(totalElapsedMillis, TimeUnit.MILLISECONDS) + "s.";
            LOG.error(errorMessage);
            throw new IllegalStateException(errorMessage);
        }
        if ((totalElapsedMillis % 5000) == 0) {
            LOG.info("Waiting for the Twill Application to start on YARN... Total elapsed time: {}s.", TimeUnit.SECONDS.convert(totalElapsedMillis, TimeUnit.MILLISECONDS));
        }
        report = controller.getResourceReport();
    }
    return report;
}