/**
 * Creates an instance.
 *
 * @param config Configuration of the yarn cluster
 * @param zkConnect ZooKeeper connection string
 * @param locationFactory Factory to create {@link Location} instances that are readable and writable by this service
 */
public YarnTwillRunnerService(YarnConfiguration config, String zkConnect, LocationFactory locationFactory) {
  this.yarnConfig = config;
  this.locationFactory = locationFactory;
  this.zkClientService = getZKClientService(zkConnect);
  this.controllers = HashBasedTable.create();
  this.serviceDelegate = new AbstractIdleService() {
    @Override
    protected void startUp() throws Exception {
      YarnTwillRunnerService.this.startUp();
    }

    @Override
    protected void shutDown() throws Exception {
      YarnTwillRunnerService.this.shutDown();
    }
  };
}
 

/** Test starting multiple instances that compete for leadership. */
@Test
public void testMultipleLeaders() throws Exception {
    final Trigger started = new Trigger();
    final AtomicInteger startCount = new AtomicInteger();
    for (int i = 0; i < 5; i++) {
        newLeaderService(1, TimeUnit.HOURS, new Supplier<Service>() {
            @Override
            public Service get() {
                return new AbstractIdleService() {
                    @Override
                    protected void startUp() throws Exception {
                        started.fire();
                        startCount.incrementAndGet();
                    }

                    @Override
                    protected void shutDown() throws Exception {
                        // Do nothing
                    }
                };
            }
        }).startAsync();
    }
    assertTrue(started.firedWithin(1, TimeUnit.MINUTES));
    // We know one service has started.  Wait a little while and verify no more services are started.
    Thread.sleep(250);
    assertTrue(startCount.get() == 1);
}