void waitForThreadOwnedOrderedConnectionState(final DistributedMember member,
      final Map<Long, Long> connectionStates)
      throws InterruptedException {
    if (Thread.interrupted()) {
      // wisest to do this before the synchronize below
      throw new InterruptedException();
    }
    List<Connection> r;
    synchronized (receivers) {
      r = new ArrayList<>(receivers);
    }
    for (Connection con : r) {
      if (!con.stopped && !con.isClosing() && !con.getOriginatedHere() && con.getPreserveOrder()
          && member.equals(con.getRemoteAddress())) {
        final Long state = connectionStates.remove(con.getUniqueId());
        if (state != null) {
          long count = state;
          while (!con.stopped && !con.isClosing() && con.getMessagesReceived() < count) {
            if (logger.isDebugEnabled()) {
              logger.debug("Waiting for connection {}/{} currently={} need={}",
                  con.getRemoteAddress(), con.getUniqueId(), con.getMessagesReceived(), count);
            }
            Thread.sleep(100);
          }
        }
      }
    }
    if (!connectionStates.isEmpty()) {
      if (logger.isDebugEnabled()) {
        StringBuilder sb = new StringBuilder(1000);
        sb.append("These connections from ");
        sb.append(member);
        sb.append("could not be located during waitForThreadOwnedOrderedConnectionState: ");
        for (Iterator<Map.Entry<Long, Long>> it = connectionStates.entrySet().iterator(); it
            .hasNext();) {
          Map.Entry<Long, Long> entry = it.next();
          sb.append(entry.getKey()).append('(').append(entry.getValue()).append(')');
          if (it.hasNext()) {
            sb.append(',');
          }
        }
        logger.debug(sb);
      }
    }
  }
wait for the given incoming connections to receive at least the associated number of messages
geode-core/src/main/java/org/apache/geode/internal/tcp/ConnectionTable.java