private void runMessagePusher() {
    try {
      final DMStats stats = owner.getConduit().getStats();
      final long threadStart = stats.startAsyncThread();
      try {
        stats.incAsyncQueues(1);
        stats.incAsyncThreads(1);

        try {
          int flushId = 0;
          while (asyncQueuingInProgress && connected) {
            if (SystemFailure.getFailure() != null) {
              // Allocate no objects here!
              Socket s = socket;
              if (s != null) {
                try {
                  logger.debug("closing socket", new Exception("closing socket"));
                  ioFilter.close(s.getChannel());
                  s.close();
                } catch (IOException e) {
                  // don't care
                }
              }
              getCheckFailure();
            }
            if (owner.getConduit().getCancelCriterion().isCancelInProgress()) {
              break;
            }
            flushId++;
            long flushStart = stats.startAsyncQueueFlush();
            try {
              long curQueuedBytes = queuedBytes;
              if (curQueuedBytes > asyncMaxQueueSize) {
                logger.warn(
                    "Queued bytes {} exceeds max of {}, asking slow receiver {} to disconnect.",
                    curQueuedBytes, asyncMaxQueueSize, remoteMember);
                stats.incAsyncQueueSizeExceeded(1);
                disconnectSlowReceiver();
                return;
              }
              SocketChannel channel = getSocket().getChannel();
              ByteBuffer bb = takeFromOutgoingQueue();
              if (bb == null) {
                if (logger.isDebugEnabled() && flushId == 1) {
                  logger.debug("P2P pusher found empty queue");
                }
                return;
              }
              writeFully(channel, bb, true, null);
              // We should not add messagesSent. The counts are increased elsewhere.
              accessed();
            } finally {
              stats.endAsyncQueueFlush(flushStart);
            }
          }
        } finally {
          // need to force this to false before doing the requestClose calls
          synchronized (outgoingQueue) {
            asyncQueuingInProgress = false;
            outgoingQueue.notifyAll();
          }
        }
      } catch (IOException ex) {
        String err = format("P2P pusher io exception for %s", this);
        if (!isSocketClosed()) {
          if (logger.isDebugEnabled() && !isIgnorableIOException(ex)) {
            logger.debug(err, ex);
          }
        }
        try {
          requestClose(err + ": " + ex);
        } catch (Exception ignore) {
        }
      } catch (CancelException ex) {
        String err = format("P2P pusher %s caught CacheClosedException: %s", this, ex);
        logger.debug(err);
        try {
          requestClose(err);
        } catch (Exception ignore) {
        }
      } catch (Exception ex) {
        owner.getConduit().getCancelCriterion().checkCancelInProgress(ex);
        if (!isSocketClosed()) {
          logger.fatal(format("P2P pusher exception: %s", ex), ex);
        }
        try {
          requestClose(format("P2P pusher exception: %s", ex));
        } catch (Exception ignore) {
        }
      } finally {
        stats.incAsyncQueueSize(-queuedBytes);
        queuedBytes = 0;
        stats.endAsyncThread(threadStart);
        stats.incAsyncThreads(-1);
        stats.incAsyncQueues(-1);
        if (logger.isDebugEnabled()) {
          logger.debug("runMessagePusher terminated id={} from {}/{}", conduitIdStr, remoteMember,
              remoteMember);
        }
      }
    } finally {
      synchronized (pusherSync) {
        pusherThread = null;
        pusherSync.notifyAll();
      }
    }
  }
have the pusher thread check for queue overflow and for idle time exceeded
geode-core/src/main/java/org/apache/geode/internal/tcp/Connection.java