public void readAck(final DirectReplyProcessor processor)
      throws SocketTimeoutException, ConnectionException {
    if (isSocketClosed()) {
      throw new ConnectionException("connection is closed");
    }
    synchronized (stateLock) {
      connectionState = STATE_READING_ACK;
    }

    boolean origSocketInUse = socketInUse;
    socketInUse = true;
    MsgReader msgReader = null;
    DMStats stats = owner.getConduit().getStats();
    final KnownVersion version = getRemoteVersion();
    try {
      msgReader = new MsgReader(this, ioFilter, version);

      ReplyMessage msg;
      int len;

      // (we have to lock here to protect between reading header and message body)
      try (final ByteBufferSharing ignored = ioFilter.getUnwrappedBuffer()) {
        Header header = msgReader.readHeader();

        if (header.getMessageType() == NORMAL_MSG_TYPE) {
          msg = (ReplyMessage) msgReader.readMessage(header);
          len = header.getMessageLength();
        } else {
          MsgDestreamer destreamer = obtainMsgDestreamer(header.getMessageId(), version);
          while (header.getMessageType() == CHUNKED_MSG_TYPE) {
            msgReader.readChunk(header, destreamer);
            header = msgReader.readHeader();
          }
          msgReader.readChunk(header, destreamer);
          msg = (ReplyMessage) destreamer.getMessage();
          releaseMsgDestreamer(header.getMessageId(), destreamer);
          len = destreamer.size();
        }
      }
      // I'd really just like to call dispatchMessage here. However,
      // that call goes through a bunch of checks that knock about
      // 10% of the performance. Since this direct-ack stuff is all
      // about performance, we'll skip those checks. Skipping them
      // should be legit, because we just sent a message so we know
      // the member is already in our view, etc.
      DistributionManager dm = owner.getDM();
      msg.setBytesRead(len);
      msg.setSender(remoteMember);
      stats.incReceivedMessages(1L);
      stats.incReceivedBytes(msg.getBytesRead());
      stats.incMessageChannelTime(msg.resetTimestamp());
      msg.process(dm, processor);
      // dispatchMessage(msg, len, false);
    } catch (SocketTimeoutException timeout) {
      throw timeout;
    } catch (IOException e) {
      final String err =
          format("ack read io exception for %s", this);
      if (!isSocketClosed()) {
        if (logger.isDebugEnabled() && !isIgnorableIOException(e)) {
          logger.debug(err, e);
        }
      }
      try {
        requestClose(err + ": " + e);
      } catch (Exception ignored) {
      }
      throw new ConnectionException(format("Unable to read direct ack because: %s", e));
    } catch (ConnectionException e) {
      owner.getConduit().getCancelCriterion().checkCancelInProgress(e);
      throw e;
    } catch (Exception e) {
      owner.getConduit().getCancelCriterion().checkCancelInProgress(e);
      if (!isSocketClosed()) {
        logger.fatal("ack read exception", e);
      }
      try {
        requestClose(format("ack read exception: %s", e));
      } catch (Exception ignored) {
      }
      throw new ConnectionException(format("Unable to read direct ack because: %s", e));
    } finally {
      stats.incProcessedMessages(1L);
      accessed();
      socketInUse = origSocketInUse;
      if (ackTimedOut) {
        logger.info("Finished waiting for reply from {}", getRemoteAddress());
        ackTimedOut = false;
      }
      if (msgReader != null) {
        msgReader.close();
      }
    }
    synchronized (stateLock) {
      connectionState = STATE_RECEIVED_ACK;
    }
  }
@throws SocketTimeoutException if wait expires.
@throws ConnectionException if ack is not received
geode-core/src/main/java/org/apache/geode/internal/tcp/Connection.java