Fix DoubleByteBuf to send large size messages in TLS mode

[nkurihar]

Motivation

Issue

When TLS is enabled, messages whose size is larger than 2^14(16384) bytes can not be sent by the following error:

java.lang.IndexOutOfBoundsException: index: 16384, length: 1000057 (expected: range(0, 1000057))
    at io.netty.buffer.AbstractByteBuf.checkIndex0(AbstractByteBuf.java:1143) ~[netty-all-4.0.40.Final.jar:4.0.40.Final]
    at io.netty.buffer.AbstractByteBuf.checkIndex(AbstractByteBuf.java:1138) ~[netty-all-4.0.40.Final.jar:4.0.40.Final]
    at io.netty.buffer.AbstractByteBuf.checkDstIndex(AbstractByteBuf.java:1157) ~[netty-all-4.0.40.Final.jar:4.0.40.Final]
    at com.yahoo.pulsar.common.api.DoubleByteBuf.getBytes(DoubleByteBuf.java:206) ~[pulsar-common-1.17.1-SNAPSHOT.jar:na]
    at com.yahoo.pulsar.common.api.DoubleByteBuf.getBytes(DoubleByteBuf.java:60) ~[pulsar-common-1.17.1-SNAPSHOT.jar:na]
    at io.netty.buffer.AbstractByteBuf.getBytes(AbstractByteBuf.java:452) ~[netty-all-4.0.40.Final.jar:4.0.40.Final]
    at com.yahoo.pulsar.common.api.DoubleByteBuf.nioBuffer(DoubleByteBuf.java:333) ~[pulsar-common-1.17.1-SNAPSHOT.jar:na]
    at com.yahoo.pulsar.common.api.DoubleByteBuf.nioBuffers(DoubleByteBuf.java:339) ~[pulsar-common-1.17.1-SNAPSHOT.jar:na]
    at com.yahoo.pulsar.common.api.DoubleByteBuf.nioBuffers(DoubleByteBuf.java:356) ~[pulsar-common-1.17.1-SNAPSHOT.jar:na]
    at io.netty.buffer.WrappedByteBuf.nioBuffers(WrappedByteBuf.java:739) ~[netty-all-4.0.40.Final.jar:4.0.40.Final]
    at io.netty.handler.ssl.SslHandler.wrap(SslHandler.java:682) ~[netty-all-4.0.40.Final.jar:na]
    at io.netty.handler.ssl.SslHandler.wrap(SslHandler.java:522) ~[netty-all-4.0.40.Final.jar:na]
    at io.netty.handler.ssl.SslHandler.flush(SslHandler.java:497) ~[netty-all-4.0.40.Final.jar:na]
    at io.netty.channel.AbstractChannelHandlerContext.invokeFlush0(AbstractChannelHandlerContext.java:780) [netty-all-4.0.40.Final.jar:4.0.40.Final]
    at io.netty.channel.AbstractChannelHandlerContext.invokeWriteAndFlush(AbstractChannelHandlerContext.java:806) [netty-all-4.0.40.Final.jar:4.0.40.Final]
    at io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:817) [netty-all-4.0.40.Final.jar:4.0.40.Final]
    at io.netty.channel.AbstractChannelHandlerContext.writeAndFlush(AbstractChannelHandlerContext.java:798) [netty-all-4.0.40.Final.jar:4.0.40.Final]
    at com.yahoo.pulsar.client.impl.ProducerImpl$WriteInEventLoopCallback.run(ProducerImpl.java:358) [pulsar-client-1.17.1-SNAPSHOT.jar:na]
    at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:408) [netty-all-4.0.40.Final.jar:4.0.40.Final]
    at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:309) [netty-all-4.0.40.Final.jar:4.0.40.Final]
    at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:140) [netty-all-4.0.40.Final.jar:4.0.40.Final]
    at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:144) [netty-all-4.0.40.Final.jar:4.0.40.Final]
    at java.lang.Thread.run(Thread.java:745) [na:1.8.0_112]

Cause

DoubleByteBuf#readableBytes returns wrong value.

Detail

DoubleByteBuf has 2 buffers (b1, b2) internally.
DoubleByteBuf#readableBytes returns b1.readableBytes() + b2.readableBytes(), where readableBytes is defined as writerIndex(default: end of the buffer) - readerIndex(default: 0).

However, since b1.readerIndex and b2.readerIndex are not updated through any methods of DoubleByteBuf, it always returns the sum of the length of b1 and b2.

In TLS, messages whose size is larger than 2^14 bytes are divided into several chunks (c.f. 6.2.1. Fragmentation on https://www.ietf.org/rfc/rfc5246.txt).

When sending the second chunk, DoubleByteBuf#readerIndex is 16384 while DoubleByteBuf#readableBytes returns the length of buffer, IndexOutOfBoundsException occurs in DoubleByteBuf#nioBuffers which attempts to read the buffer from 16384 to 16384 + length.

Modifications

• Delete DoubleByteBuf#readableBytes in order to use AbstractByteBuf#readableBytes instead
• Added TlsProducerConsumerTest to verify that messages whose size is larger than 2^14 bytes can be sent.
• Fixed typo in AuthenticatedProducerConsumerTest

Note:
In this PR I just deleted DoubleByteBuf#readableBytes.
Another approach is to update b1.readerIndex and b2.readerIndex when DoubleByteBuf#readerIndex is updated, but it seems unnecessary for now.
Should we do so?

Result

Messages whose size is larger than 2^14 bytes can be sent when TLS is enabled.

[merlimat]

@nkurihar The change looks good. I just have a couple of questions:

• What are the values of b1.readableBytes() and b2.readableBytes() ?
• Why is this happening only for TLS and not regular messages?
• Can you add a test method in DoubleByteBuf to demonstrate this issue just with the buffer itself?

[nkurihar]

@merlimat

What are the values of b1.readableBytes() and b2.readableBytes() ?

They are defined as b1(b2).writerIndex - b1(b2).readerIndex.
In this case, readerIndex and writerIndex are the start and the end of the buffer, respectively.
Since no method in DoubleByteBuf updates readerIndex and writerIndex, b1(b2).readableBytes() always returns the length of the buffer.

Why is this happening only for TLS and not regular messages?

If TLS is disabled, messages larger than 16384 bytes can be sent at one time.
However, if TLS is enabled, such messages are divided into chunks and sent at more than one time.

Can you add a test method in DoubleByteBuf to demonstrate this issue just with the buffer itself?

I added a test code for DoubleByteBuf#readableBytes.

[merlimat]

Got it, thanks