### Question:
RunningDatanodeState implements DatanodeState { @Override public DatanodeStateMachine.DatanodeStates await(long duration, TimeUnit timeUnit) throws InterruptedException { int count = connectionManager.getValues().size(); int returned = 0; long durationMS = timeUnit.toMillis(duration); long timeLeft = durationMS; long startTime = Time.monotonicNow(); List<Future<EndPointStates>> results = new LinkedList<>(); while (returned < count && timeLeft > 0) { Future<EndPointStates> result = ecs.poll(timeLeft, TimeUnit.MILLISECONDS); if (result != null) { results.add(result); returned++; } timeLeft = durationMS - (Time.monotonicNow() - startTime); } return computeNextContainerState(results); } RunningDatanodeState(ConfigurationSource conf,
      SCMConnectionManager connectionManager,
      StateContext context); @Override void onEnter(); @Override void onExit(); @Override void execute(ExecutorService executor); @VisibleForTesting void setExecutorCompletionService(ExecutorCompletionService e); @Override DatanodeStateMachine.DatanodeStates await(long duration, TimeUnit timeUnit);  }### Answer:
@Test public void testAwait() throws InterruptedException { SCMConnectionManager connectionManager = Mockito.mock(SCMConnectionManager.class); List<EndpointStateMachine> stateMachines = new ArrayList<>(); when(connectionManager.getValues()).thenReturn(stateMachines); RunningDatanodeState state = new RunningDatanodeState(null, connectionManager, null); int threadPoolSize = 2; ExecutorService executorService = Executors.newFixedThreadPool( threadPoolSize); ExecutorCompletionService ecs = new ExecutorCompletionService<>(executorService); state.setExecutorCompletionService(ecs); for (int i = 0; i < threadPoolSize; i++) { stateMachines.add(new EndpointStateMachine(null, null, null)); } CompletableFuture<EndpointStateMachine.EndPointStates> futureOne = new CompletableFuture<>(); for (int i = 0; i < threadPoolSize; i++) { ecs.submit(() -> futureOne.get()); } long startTime = Time.monotonicNow(); state.await(500, TimeUnit.MILLISECONDS); long endTime = Time.monotonicNow(); Assert.assertTrue((endTime - startTime) >= 500); futureOne.complete(SHUTDOWN); CompletableFuture<EndpointStateMachine.EndPointStates> futureTwo = new CompletableFuture<>(); for (int i = 0; i < threadPoolSize; i++) { ecs.submit(() -> futureTwo.get()); } futureTwo.complete(SHUTDOWN); startTime = Time.monotonicNow(); state.await(500, TimeUnit.MILLISECONDS); endTime = Time.monotonicNow(); Assert.assertTrue((endTime - startTime) < 500); executorService.shutdown(); }