Large message deliveries with TLS and NIO enabled results in a "buffer closed" exception in SslEngineHelper

[YuryAndr]

Hello!

There was an issue #307 .

And it looks like it reproduced again.

Versions:
Server - RabbitMQ 3.8.11

Client -

<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.13.0</version>
</dependency>

Steps to reproduce

• SSL and NIO are enabled
• Send a large message to queue
• Start consumer of the queue

Result
Connection is closed with error

2021/07/28 19:50:17.240 ERROR [rabbitmq-connection-shutdown-amqp://guest@127.0.0.1:5671/test] c.r.c.i.ForgivingExceptionHandler - An unexpected connection driver error occurred 
javax.net.ssl.SSLException: Unrecognized record version (D)TLS-0.0 , plaintext connection?
	at java.base/sun.security.ssl.SSLEngineInputRecord.bytesInCompletePacket(SSLEngineInputRecord.java:98)
	at java.base/sun.security.ssl.SSLEngineInputRecord.bytesInCompletePacket(SSLEngineInputRecord.java:64)
	at java.base/sun.security.ssl.SSLEngineImpl.readRecord(SSLEngineImpl.java:548)
	at java.base/sun.security.ssl.SSLEngineImpl.unwrap(SSLEngineImpl.java:443)
	at java.base/sun.security.ssl.SSLEngineImpl.unwrap(SSLEngineImpl.java:422)
	at java.base/javax.net.ssl.SSLEngine.unwrap(SSLEngine.java:634)
	at com.rabbitmq.client.impl.nio.SslEngineFrameBuilder.somethingToRead(SslEngineFrameBuilder.java:49)
	at com.rabbitmq.client.impl.nio.FrameBuilder.readFrame(FrameBuilder.java:71)
	at com.rabbitmq.client.impl.nio.NioLoop.run(NioLoop.java:156)
	at java.base/java.lang.Thread.run(Thread.java:834)
2021/07/28 19:50:17.241 ERROR [rabbitmq-connection-shutdown-amqp://guest@127.0.0.1:5671/test] c.r.c.i.ForgivingExceptionHandler - An unexpected connection driver error occurred 
javax.net.ssl.SSLException: Buffer closed
	at com.rabbitmq.client.impl.nio.SslEngineHelper.write(SslEngineHelper.java:176)
	at com.rabbitmq.client.impl.nio.SslEngineByteBufferOutputStream.doFlush(SslEngineByteBufferOutputStream.java:59)
	at com.rabbitmq.client.impl.nio.SslEngineByteBufferOutputStream.flush(SslEngineByteBufferOutputStream.java:53)
	at java.base/java.io.DataOutputStream.flush(DataOutputStream.java:123)
	at com.rabbitmq.client.impl.nio.NioLoop.run(NioLoop.java:244)
	at java.base/java.lang.Thread.run(Thread.java:834)

PS when using rabbit client without nio - it works fine

[michaelklishin]

How large is large? Can you put together an executable example that reproduces against 3.8.19 or 3.9.0?

[YuryAndr]

It looks like it depends on network delay. It can be reproduced with message of 15_000 bytes for remote AWS RabbitMQ broker.

Also i have configured local setup with RabbitMQ 3.8.19 and reproduces error with 70_000 bytes.

AWS RabbitMQ broker supports 3.8.11 version.
Next example shows such error.

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
import java.nio.charset.StandardCharsets;


public class Send {

    private final static String QUEUE_NAME = "hello";

    public static void main(String[] argv) throws Exception {
        startConsumer();

        ConnectionFactory factory = getConnectionFactory();
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);

            String message = ".".repeat(70_000);

            channel.basicPublish("", QUEUE_NAME, null, message.getBytes(StandardCharsets.UTF_8));
            System.out.println(" [x] Sent '" + message.length() + "'");
        }
    }

    private static void startConsumer() throws Exception {
        ConnectionFactory factory = getConnectionFactory();

        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println(" [x] Received " + message.length());
        };
        channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { });
    }

    private static ConnectionFactory getConnectionFactory() throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("your_host");
        factory.setPort(5671);
        factory.useSslProtocol();
        factory.setUsername("your_user_name");
        factory.setPassword("your_password");
        factory.useNio();
        return factory;
    }
}

[michaelklishin]

Our team could reproduce 👍

[acogoluegnes]

Thanks for the example. I would suggest to use a CountDownLatch to block until the message is received (or time out), this avoids race conditions where the program could exit before the message actually arrives. Disabling automatic connection recovery can be useful.

I managed to reproduce with the example, but not locally, I had to use a remote broker. The NioTlsUnverifiedConnection test in the test suite can also reproduce, but again with a remote broker. I provided a fix #702.

[YuryAndr]

@acogoluegnes Thank you for quick feedback!

I suppose that PR fix #702 doesn't take into account cipherIn.clear() before next chunk reading at SocketChannelFrameHandlerState.

I created PR to handle this
https://github.com/rabbitmq/rabbitmq-java-client/pull/701/files

[michaelklishin]

We may introduce follow-up changes to #701 but I assume it gets us most of the way. Kudos to @YuryAndr for
thoroughly investigating this and contributing a fix.