private void handleAckTimeout(long ackTimeout, long ackSATimeout, Connection c,
      DirectReplyProcessor processor) throws ConnectionException {
    Set activeMembers = dm.getDistributionManagerIds();

    // Increment the stat
    dm.getStats().incReplyTimeouts();

    // an alert that will show up in the console
    {
      String msg =
          "%s seconds have elapsed while waiting for reply from %s on %s whose current membership list is: [%s]";
      final Object[] msgArgs = new Object[] {ackTimeout / 1000, c.getRemoteAddress(),
          dm.getId(), activeMembers};
      logger.warn(String.format(msg, msgArgs));
      msgArgs[3] = "(omitted)";
      Breadcrumbs.setProblem(msg, msgArgs);

      if (ReplyProcessor21.THROW_EXCEPTION_ON_TIMEOUT) {
        // init the cause to be a TimeoutException so catchers can determine cause
        TimeoutException cause =
            new TimeoutException("Timed out waiting for ACKS.");
        throw new InternalGemFireException(String.format(msg, msgArgs), cause);
      }
    }

    if (activeMembers.contains(c.getRemoteAddress())) {
      // wait for ack-severe-alert-threshold period first, then wait forever
      if (ackSATimeout > 0) {
        try {
          c.readAck(processor);
          return;
        } catch (SocketTimeoutException e) {
          Object[] args = new Object[] {(ackSATimeout + ackTimeout) / 1000,
              c.getRemoteAddress(), dm.getId(), activeMembers};
          logger.fatal(
              "{} seconds have elapsed while waiting for reply from {} on {} whose currentFull membership list is: [{}]",
              args);
        }
      }
      try {
        c.readAck(processor);
      } catch (SocketTimeoutException ex) {
        // this can never happen when called with timeout of 0
        logger.error(String.format("Unexpected timeout while waiting for ack from %s",
            c.getRemoteAddress()),
            ex);
      }
    } else {
      logger.warn("View no longer has {} as an active member, so we will no longer wait for it.",
          c.getRemoteAddress());
      processor.memberDeparted(getDM(), c.getRemoteAddress(), true);
    }
  }
@param ackTimeout ack wait threshold
@param ackSATimeout severe alert threshold
geode-core/src/main/java/org/apache/geode/distributed/internal/direct/DirectChannel.java