Continuations (requests) are not discarded on timeouts

[wheezil]

I am starting to use the RpcClient class, and examining the source it apparently can leak by adding entries to _continuationMap that are never removed if there is no reply. If doCall() throws TimeoutException, there is no reason to leave this entry in the map as it will never be used, so it should catch TimeoutException and remove the entry.

• RabbitMQ version: 5.3.0 (but issue remains in master code)

Relevant code follows. The call to k.uninterruptibleGet() can throw TimeoutException. We can catch that and remove the entry before re-throwing it.

public Response doCall(AMQP.BasicProperties props, byte[] message, int timeout)
    throws IOException, ShutdownSignalException, TimeoutException {
    checkConsumer();
    BlockingCell<Object> k = new BlockingCell<Object>();
    synchronized (_continuationMap) {
        _correlationId++;
        String replyId = "" + _correlationId;
        props = ((props==null) ? new AMQP.BasicProperties.Builder() : props.builder())
            .correlationId(replyId).replyTo(_replyTo).build();
        _continuationMap.put(replyId, k);
    }
    publish(props, message);
    Object reply = k.uninterruptibleGet(timeout);
    if (reply instanceof ShutdownSignalException) {
        ShutdownSignalException sig = (ShutdownSignalException) reply;
        ShutdownSignalException wrapper =
            new ShutdownSignalException(sig.isHardError(),
                                        sig.isInitiatedByApplication(),
                                        sig.getReason(),
                                        sig.getReference());
        wrapper.initCause(sig);
        throw wrapper;
    } else {
        return (Response) reply;
    }
}

I've already posted this to the rabbitmq mailing list, and got the thumbs-up to submit a patch, so here goes! Please forgive my newbie status; I am still struggling getting this to build.