Do not preallocate bytes for channel buffer (#31400)

Currently, when we open a new channel, we pass it an
InboundChannelBuffer. The channel buffer is preallocated a single 16kb
page. However, there is no guarantee that this channel will be read from
anytime soon. Instead, this commit does not preallocate that page. That
page will be allocated when we receive a read event.

libs/nio/src/main/java/org/elasticsearch/nio/InboundChannelBuffer.java

@@ -58,7 +58,6 @@ public InboundChannelBuffer(Supplier<Page> pageSupplier) {
         this.pageSupplier = pageSupplier;
         this.pages = new ArrayDeque<>();
         this.capacity = PAGE_SIZE * pages.size();
-        ensureCapacity(PAGE_SIZE);
     }
 
     public static InboundChannelBuffer allocatingInstance() {

libs/nio/src/test/java/org/elasticsearch/nio/InboundChannelBufferTests.java

@@ -34,16 +34,20 @@ public class InboundChannelBufferTests extends ESTestCase {
         new InboundChannelBuffer.Page(ByteBuffer.allocate(BigArrays.BYTE_PAGE_SIZE), () -> {
         });
 
-    public void testNewBufferHasSinglePage() {
+    public void testNewBufferNoPages() {
         InboundChannelBuffer channelBuffer = new InboundChannelBuffer(defaultPageSupplier);
 
-        assertEquals(PAGE_SIZE, channelBuffer.getCapacity());
-        assertEquals(PAGE_SIZE, channelBuffer.getRemaining());
+        assertEquals(0, channelBuffer.getCapacity());
+        assertEquals(0, channelBuffer.getRemaining());
         assertEquals(0, channelBuffer.getIndex());
     }
 
     public void testExpandCapacity() {
         InboundChannelBuffer channelBuffer = new InboundChannelBuffer(defaultPageSupplier);
+        assertEquals(0, channelBuffer.getCapacity());
+        assertEquals(0, channelBuffer.getRemaining());
+
+        channelBuffer.ensureCapacity(PAGE_SIZE);
 
         assertEquals(PAGE_SIZE, channelBuffer.getCapacity());
         assertEquals(PAGE_SIZE, channelBuffer.getRemaining());
@@ -56,6 +60,7 @@ public void testExpandCapacity() {
 
     public void testExpandCapacityMultiplePages() {
         InboundChannelBuffer channelBuffer = new InboundChannelBuffer(defaultPageSupplier);
+        channelBuffer.ensureCapacity(PAGE_SIZE);
 
         assertEquals(PAGE_SIZE, channelBuffer.getCapacity());
 
@@ -68,6 +73,7 @@ public void testExpandCapacityMultiplePages() {
 
     public void testExpandCapacityRespectsOffset() {
         InboundChannelBuffer channelBuffer = new InboundChannelBuffer(defaultPageSupplier);
+        channelBuffer.ensureCapacity(PAGE_SIZE);
 
         assertEquals(PAGE_SIZE, channelBuffer.getCapacity());
         assertEquals(PAGE_SIZE, channelBuffer.getRemaining());
@@ -87,6 +93,7 @@ public void testExpandCapacityRespectsOffset() {
 
     public void testIncrementIndex() {
         InboundChannelBuffer channelBuffer = new InboundChannelBuffer(defaultPageSupplier);
+        channelBuffer.ensureCapacity(PAGE_SIZE);
 
         assertEquals(0, channelBuffer.getIndex());
         assertEquals(PAGE_SIZE, channelBuffer.getRemaining());
@@ -99,6 +106,7 @@ public void testIncrementIndex() {
 
     public void testIncrementIndexWithOffset() {
         InboundChannelBuffer channelBuffer = new InboundChannelBuffer(defaultPageSupplier);
+        channelBuffer.ensureCapacity(PAGE_SIZE);
 
         assertEquals(0, channelBuffer.getIndex());
         assertEquals(PAGE_SIZE, channelBuffer.getRemaining());

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/transport/nio/SSLDriver.java

@@ -365,6 +365,7 @@ private void handshake() throws SSLException {
 
         @Override
         public void read(InboundChannelBuffer buffer) throws SSLException {
+            ensureApplicationBufferSize(buffer);
             boolean continueUnwrap = true;
             while (continueUnwrap && networkReadBuffer.position() > 0) {
                 networkReadBuffer.flip();