private static boolean awaitSideInputTasks(Map<TaskName, CountDownLatch> sideInputTaskLatches) throws InterruptedException {
    long endTime = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(SIDE_INPUT_CHECK_TIMEOUT_SECONDS);
    for (CountDownLatch latch : sideInputTaskLatches.values()) {
      long remainingMillisToWait = endTime - System.currentTimeMillis();
      if (remainingMillisToWait <= 0 || !latch.await(remainingMillisToWait, TimeUnit.MILLISECONDS)) {
        return false;
      }
    }
    return true;
  }
Waits for all side input tasks to catch up until a timeout.

@return False if waiting on any latch timed out, true otherwise

@throws InterruptedException if waiting any of the latches is interrupted
samza-core/src/main/scala/org/apache/samza/storage/SideInputsManager.java