@SuppressWarnings("unchecked")
    @Override
    public void onClosed(final String remoteAddress, final Connection conn) {
        final Set<PeerPair> pairs = (Set<PeerPair>) conn.getAttribute(PAIR_ATTR);
        if (pairs != null && !pairs.isEmpty()) {
            // Clear request contexts when connection disconnected.
            for (final Map.Entry<String, ConcurrentMap<PeerPair, PeerRequestContext>> entry : this.peerRequestContexts
                .entrySet()) {
                final ConcurrentMap<PeerPair, PeerRequestContext> groupCtxs = entry.getValue();
                synchronized (Utils.withLockObject(groupCtxs)) {
                    for (PeerPair pair : pairs) {
                        final PeerRequestContext ctx = groupCtxs.remove(pair);
                        if (ctx != null) {
                            ctx.destroy();
                        }
                    }
                }
            }
        } else {
            LOG.info("Connection disconnected: {}", remoteAddress);
        }
    }
@Test
    public void testOnClosed() {
        mockNode();
        final AppendEntriesRequestProcessor processor = (AppendEntriesRequestProcessor) newProcessor();

        PeerPair pair = processor.pairOf(this.peerIdStr, this.serverId);
        final PeerRequestContext ctx = processor.getOrCreatePeerRequestContext(this.groupId, pair, this.conn);
        assertNotNull(ctx);
        assertSame(ctx, processor.getPeerRequestContext(this.groupId, pair));
        assertSame(ctx, processor.getOrCreatePeerRequestContext(this.groupId, pair, this.conn));

        processor.onClosed(null, this.conn);
        assertNull(processor.getPeerRequestContext(this.groupId, pair));
        assertNotSame(ctx, processor.getOrCreatePeerRequestContext(this.groupId, pair, this.conn));
    }