void sendSequenceResponse(final String groupId, final PeerPair pair, final int seq, final RpcContext rpcCtx,
                              final Message msg) {
        final PeerRequestContext ctx = getPeerRequestContext(groupId, pair);
        if (ctx == null) {
            // the context was destroyed, so the response can be ignored.
            return;
        }
        final PriorityQueue<SequenceMessage> respQueue = ctx.responseQueue;
        assert (respQueue != null);

        synchronized (Utils.withLockObject(respQueue)) {
            respQueue.add(new SequenceMessage(rpcCtx, msg, seq));

            if (!ctx.hasTooManyPendingResponses()) {
                while (!respQueue.isEmpty()) {
                    final SequenceMessage queuedPipelinedResponse = respQueue.peek();

                    if (queuedPipelinedResponse.sequence != ctx.getNextRequiredSequence()) {
                        // sequence mismatch, waiting for next response.
                        break;
                    }
                    respQueue.remove();
                    try {
                        queuedPipelinedResponse.sendResponse();
                    } finally {
                        ctx.getAndIncrementNextRequiredSequence();
                    }
                }
            } else {
                final Connection connection = rpcCtx.getConnection();
                LOG.warn("Closed connection to peer {}/{}, because of too many pending responses, queued={}, max={}",
                    ctx.groupId, pair, respQueue.size(), ctx.maxPendingResponses);
                connection.close();
                // Close the connection if there are too many pending responses in queue.
                removePeerRequestContext(groupId, pair);
            }
        }
    }
@Test
    public void testSendSequenceResponse() {
        mockNode();
        final AppendEntriesRequestProcessor processor = (AppendEntriesRequestProcessor) newProcessor();
        final PeerPair pair = processor.pairOf(this.peerIdStr, this.serverId);
        processor.getOrCreatePeerRequestContext(this.groupId, pair, this.conn);
        final PingRequest msg = TestUtils.createPingRequest();
        final RpcContext asyncContext = Mockito.mock(RpcContext.class);
        processor.sendSequenceResponse(this.groupId, pair, 1, asyncContext, msg);
        Mockito.verify(asyncContext, Mockito.never()).sendResponse(msg);

        processor.sendSequenceResponse(this.groupId, pair, 0, asyncContext, msg);
        Mockito.verify(asyncContext, Mockito.times(2)).sendResponse(msg);
    }