code: protected MessageTuple getExpectedMessage(MessageType type,
                                              String source,
                                              int timeoutMs)
        throws TimeoutException, InterruptedException {
      int startTime = (int) (System.nanoTime() / 1000000);
      // Waits until the expected message is received.
      while (true) {
        MessageTuple tuple = getMessage(timeoutMs);
        String from = tuple.getServerId();
        if (tuple.getMessage().getType() == type &&
            (source == null || source.equals(from))) {
          // Return message only if it's expected type and expected source.
          return tuple;
        } else {
          int curTime = (int) (System.nanoTime() / 1000000);
          if (curTime - startTime >= timeoutMs) {
            throw new TimeoutException("Timeout in getExpectedMessage.");
          }
          if (LOG.isDebugEnabled()) {
            LOG.debug("Got an unexpected message from {}: {}",
                      tuple.getServerId(),
                      TextFormat.shortDebugString(tuple.getMessage()));
          }
        }
      }
    }
docs: Takes the expected message from message queue. This method will be blocked
  until the expected message from the expected source becomes
  available or timeout is detected.
  
  @param type the expected message type.
  @param source the expected source of the message, or null if the source
  doesn't matter.
  @param timeoutMs the timeout for blocking, in millisecond.
  @return filtered message tuple.
  @throws InterruptedException in case of interrupt on blocking.
  @throws TimeoutException if timeout happens during blocking.