### Question:
ConnectorTraverser extends ConnectorScheduler<IndexingConnectorContext> { @Override public synchronized void start() { super.start(); ConnectorSchedule schedule = getConnectorSchedule(); if (schedule.isRunOnce()) { traversers = Collections.emptyList(); } else { List<TraverserWorker> traverserWorkers = new ArrayList<>(); for (TraverserConfiguration configuration : context.getTraverserConfiguration()) { TraverserWorker traverser = TraverserWorkerManager.newWorker( configuration, context.getIndexingService(), getBackgroundExecutor()); traverserWorkers.add(traverser); Runnable pollQueueRunnable = new BackgroundRunnable( new OneAtATimeRunnable(() -> traverser.poll(), traverser.getName())); getScheduledExecutor().scheduleAtFixedRate( pollQueueRunnable, 0, schedule.getPollQueueIntervalSecs(), TimeUnit.SECONDS); } traversers = Collections.unmodifiableList(traverserWorkers); } } private  ConnectorTraverser(Builder builder); @Override synchronized void start(); @Override synchronized void stop();  }### Answer:
@Test public void testStartAndRunOnce() throws Exception { setupConfig(Collections.singletonMap("connector.runOnce", "true")); CountDownLatch shutdown = new CountDownLatch(1); IndexingConnectorContext context = new IndexingConnectorContextImpl.Builder().setIndexingService(mockIndexingService).build(); NothingConnector connector = new NothingConnector(1); ConnectorTraverser traverser = new ConnectorTraverser.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); ConnectorTraverser traverser = new ConnectorTraverser.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)); }