private boolean addToQueue(ByteBuffer buffer, DistributionMessage msg, boolean force)
      throws ConnectionException {
    final DMStats stats = owner.getConduit().getStats();
    long start = DistributionStats.getStatTime();
    try {
      ConflationKey ck = null;
      if (msg != null) {
        ck = msg.getConflationKey();
      }
      Object objToQueue = null;
      // if we can conflate delay the copy to see if we can reuse an already allocated buffer.
      final int newBytes = buffer.remaining();
      final int origBufferPos = buffer.position();
      if (ck == null || !ck.allowsConflation()) {
        // do this outside of sync for multi thread perf
        ByteBuffer newbb = ByteBuffer.allocate(newBytes);
        newbb.put(buffer);
        newbb.flip();
        objToQueue = newbb;
      }
      synchronized (outgoingQueue) {
        if (disconnectRequested) {
          buffer.position(origBufferPos);
          // we have given up so just drop this message.
          throw new ConnectionException(format("Forced disconnect sent to %s", remoteMember));
        }
        if (!force && !asyncQueuingInProgress) {
          // reset buffer since we will be sending it
          buffer.position(origBufferPos);
          // the pusher emptied the queue so don't add since we are not forced to.
          return false;
        }
        boolean didConflation = false;
        if (ck != null) {
          if (ck.allowsConflation()) {
            objToQueue = ck;
            ConflationKey oldValue = conflatedKeys.put(ck, ck);
            if (oldValue != null) {
              ByteBuffer oldBuffer = oldValue.getBuffer();
              // need to always do this to allow old buffer to be gc'd
              oldValue.setBuffer(null);

              // remove the conflated key from current spot in queue

              // Note we no longer remove from the queue because the search
              // can be expensive on large queues. Instead we just wait for
              // the queue removal code to find the oldValue and ignore it since
              // its buffer is null

              // We do a quick check of the last thing in the queue
              // and if it has the same identity of our last thing then
              // remove it

              if (outgoingQueue.getLast() == oldValue) {
                outgoingQueue.removeLast();
              }
              int oldBytes = oldBuffer.remaining();
              queuedBytes -= oldBytes;
              stats.incAsyncQueueSize(-oldBytes);
              stats.incAsyncConflatedMsgs();
              didConflation = true;
              if (oldBuffer.capacity() >= newBytes) {
                // copy new buffer into oldBuffer
                oldBuffer.clear();
                oldBuffer.put(buffer);
                oldBuffer.flip();
                ck.setBuffer(oldBuffer);
              } else {
                // old buffer was not large enough
                ByteBuffer newbb = ByteBuffer.allocate(newBytes);
                newbb.put(buffer);
                newbb.flip();
                ck.setBuffer(newbb);
              }
            } else {
              // no old buffer so need to allocate one
              ByteBuffer newbb = ByteBuffer.allocate(newBytes);
              newbb.put(buffer);
              newbb.flip();
              ck.setBuffer(newbb);
            }
          } else {
            // just forget about having a conflatable operation
            conflatedKeys.remove(ck);
          }
        }

        long newQueueSize = newBytes + queuedBytes;
        if (newQueueSize > asyncMaxQueueSize) {
          logger.warn("Queued bytes {} exceeds max of {}, asking slow receiver {} to disconnect.",
              newQueueSize, asyncMaxQueueSize, remoteMember);
          stats.incAsyncQueueSizeExceeded(1);
          disconnectSlowReceiver();
          // reset buffer since we will be sending it
          buffer.position(origBufferPos);
          return false;
        }

        outgoingQueue.addLast(objToQueue);
        queuedBytes += newBytes;
        stats.incAsyncQueueSize(newBytes);
        if (!didConflation) {
          stats.incAsyncQueuedMsgs();
        }
        return true;
      }
    } finally {
      if (DistributionStats.enableClockStats) {
        stats.incAsyncQueueAddTime(DistributionStats.getStatTime() - start);
      }
    }
  }
ack-wait-threshold and ack-severe-alert-threshold processing
geode-core/src/main/java/org/apache/geode/internal/tcp/Connection.java