@Override
public void startServices() {
  writeLock.lock();
  try {
    if (!servicesInited.get()) {
      initServices();
    }
    for (AbstractService srvc : services) {
      if (LOG.isDebugEnabled()) {
        LOG.debug("starting service : " + srvc.getName()
            + ", for vertex: " + getName());
      }
      srvc.start();
    }
  } finally {
    writeLock.unlock();
  }
}
 

@Override
public void serviceStop() {
  synchronized(this) {
    this.stopEventHandling = true;
    if (eventHandlingThread != null)
      eventHandlingThread.interrupt();
  }
  if (taskScheduler != null) {
    ((AbstractService)taskScheduler).stop();
  }
}
 

@Override
public void serviceStart() {
  TaskSchedulerService taskSchedulerReal = createTaskScheduler("host", 0, "",
    appContext);
  // Init the service so that reuse configuration is picked up.
  ((AbstractService)taskSchedulerReal).init(getConfig());
  ((AbstractService)taskSchedulerReal).start();
  taskScheduler = spy(taskSchedulerReal);
}
 

@Override
public void initServices() {
  if (servicesInited.get()) {
    if (LOG.isDebugEnabled()) {
      LOG.debug("Skipping Initing services for vertex because already"
          + " Initialized, name=" + this.vertexName);
    }
    return;
  }
  writeLock.lock();
  try {
    List<AbstractService> servicesToAdd = new ArrayList<>();
    if (isSpeculationEnabled()) {
      // Initialize the speculator
      if (LOG.isDebugEnabled()) {
        LOG.debug(
            "Initing service vertex speculator, name=" + this.vertexName);
      }
      speculator = new LegacySpeculator(vertexConf, getAppContext(), this);
      speculator.init(vertexConf);
      servicesToAdd.add(speculator);
    }
    services = Collections.synchronizedList(servicesToAdd);
    servicesInited.set(true);
  } finally {
    writeLock.unlock();
  }
  if (LOG.isDebugEnabled()) {
    LOG.debug("Initing service vertex, name=" + this.vertexName);
  }
}
 

protected AbstractService createStagingDirCleaningService() {
  return new StagingDirCleaningService();
}
 

private void register(ServiceStateChangeListener l) {
  AbstractService.registerGlobalListener(l);
}
 

private boolean unregister(ServiceStateChangeListener l) {
  return AbstractService.unregisterGlobalListener(l);
}
 

/**
 * After every test case reset the list of global listeners.
 */
@After
public void cleanup() {
  AbstractService.resetGlobalListeners();
}
 

protected AbstractService createStagingDirCleaningService() {
  return new StagingDirCleaningService();
}
 

private void register(ServiceStateChangeListener l) {
  AbstractService.registerGlobalListener(l);
}
 

private boolean unregister(ServiceStateChangeListener l) {
  return AbstractService.unregisterGlobalListener(l);
}
 

/**
 * After every test case reset the list of global listeners.
 */
@After
public void cleanup() {
  AbstractService.resetGlobalListeners();
}
 

@Override
public AbstractService getSpeculator() { return speculator; }
 

AbstractService getSpeculator();