### Question:
ConnectorScheduler { public synchronized void start() { checkNotNull(context); boolean success = isRunning.compareAndSet(false, true); checkState(success, "Connector traverser already started."); scheduleExecutor = Executors.newSingleThreadScheduledExecutor( new ThreadFactoryBuilder().setDaemon(false).setNameFormat("schedule").build()); backgroundExecutor = Executors.newCachedThreadPool( new ThreadFactoryBuilder().setDaemon(false).setNameFormat("background").build()); ConnectorSchedule traversalSchedule = new ConnectorSchedule(); if (traversalSchedule.isRunOnce()) { startToRunOnce(traversalSchedule); } else { startToRunContinuously(traversalSchedule); } Runnable loggingStatsRunnable = new BackgroundRunnable( new OneAtATimeRunnable( () -> logger.info(StatsManager.getInstance().printStats()), "StatsLog")); scheduleExecutor.scheduleAtFixedRate(loggingStatsRunnable, 1, 5, TimeUnit.MINUTES); } protected  ConnectorScheduler(AbstractBuilder<? extends AbstractBuilder, T> builder); synchronized void start(); synchronized void stop(); boolean isStarted();  }### Answer:
@Test public void testStartAndRunOnce() throws Exception { setupConfig(Collections.singletonMap("connector.runOnce", "true")); CountDownLatch shutdown = new CountDownLatch(1); ConnectorContext context = new ConnectorContextImpl.Builder().build(); NothingConnector connector = new NothingConnector(1); ConnectorScheduler<ConnectorContext> traverser = new ConnectorScheduler.Builder() .setConnector(connector) .setContext(context) .setShutdownHolder(() -> shutdown.countDown()) .build(); traverser.start(); assertTrue(connector.latch.await(30, TimeUnit.SECONDS)); assertTrue(shutdown.await(30, TimeUnit.SECONDS)); }
@Test public void testStartAlreadyStarted() throws Exception { setupConfig(Collections.emptyMap()); NothingConnector connector = new NothingConnector(1); ConnectorScheduler<ConnectorContext> traverser = new ConnectorScheduler.Builder() .setConnector(connector) .setContext(getContextWithExceptionHandler(-1)) .build(); traverser.start(); thrown.expect(IllegalStateException.class); traverser.start(); }
@Test public void testTraverseLongRunning() throws Exception { CountDownLatch alreadyRunningLatch = new CountDownLatch(2); Runnable alreadyRunningRunnable = () -> { assertThat(alreadyRunningLatch.getCount(), not(equalTo(0))); alreadyRunningLatch.countDown(); }; CountDownLatch longRunningLatch = new CountDownLatch(2); CountDownLatch firstRunLatch = new CountDownLatch(1); Runnable longRunningRunnable = () -> { try { assertTrue(alreadyRunningLatch.await(30, TimeUnit.SECONDS)); longRunningLatch.countDown(); firstRunLatch.countDown(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new RuntimeException(e); } }; OneAtATimeRunnable subject = new OneAtATimeRunnable(longRunningRunnable, alreadyRunningRunnable); Thread thread1 = new Thread(subject); Thread thread2 = new Thread(subject); Thread thread3 = new Thread(subject); Thread thread4 = new Thread(subject); thread1.start(); thread2.start(); thread3.start(); assertTrue(firstRunLatch.await(30, TimeUnit.SECONDS)); thread4.start(); assertTrue(longRunningLatch.await(30, TimeUnit.SECONDS)); assertTrue(alreadyRunningLatch.await(0, TimeUnit.SECONDS)); }