Add Ref Count Assertion to Page (elastic#69599)

Added this assertion to have an easier time debugging
work on elastic#67502 and found that we were accessing `refcount == 0`
bytes in the `SSLOutboundBuffer` so I fixed that buffer to not
keep references to released pages.

Author: original-brownbear

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

@@ -22,15 +22,12 @@ public class Page implements Closeable {
     // released.
     private final RefCountedCloseable refCountedCloseable;
 
-    public Page(ByteBuffer byteBuffer) {
-        this(byteBuffer, () -> {});
-    }
-
     public Page(ByteBuffer byteBuffer, Runnable closeable) {
         this(byteBuffer, new RefCountedCloseable(closeable));
     }
 
     private Page(ByteBuffer byteBuffer, RefCountedCloseable refCountedCloseable) {
+        assert refCountedCloseable.refCount() > 0;
         this.byteBuffer = byteBuffer;
         this.refCountedCloseable = refCountedCloseable;
     }
@@ -53,6 +50,7 @@ public Page duplicate() {
      * @return the byte buffer
      */
     public ByteBuffer byteBuffer() {
+        assert refCountedCloseable.refCount() > 0;
         return byteBuffer;
     }
 

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

@@ -6,13 +6,14 @@
  */
 package org.elasticsearch.xpack.security.transport.nio;
 
+import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.core.internal.io.IOUtils;
 import org.elasticsearch.nio.FlushOperation;
 import org.elasticsearch.nio.Page;
 
 import java.nio.ByteBuffer;
 import java.util.ArrayDeque;
-import java.util.function.BiConsumer;
 import java.util.function.IntFunction;
 
 public class SSLOutboundBuffer implements AutoCloseable {
@@ -49,10 +50,6 @@ ByteBuffer nextWriteBuffer(int networkBufferSize) {
     }
 
     FlushOperation buildNetworkFlushOperation() {
-        return buildNetworkFlushOperation((r, e) -> {});
-    }
-
-    FlushOperation buildNetworkFlushOperation(BiConsumer<Void, Exception> listener) {
         int pageCount = pages.size();
         ByteBuffer[] byteBuffers = new ByteBuffer[pageCount];
         Page[] pagesToClose = new Page[pageCount];
@@ -63,8 +60,15 @@ FlushOperation buildNetworkFlushOperation(BiConsumer<Void, Exception> listener)
         }
 
         return new FlushOperation(byteBuffers, (r, e) -> {
-            IOUtils.closeWhileHandlingException(pagesToClose);
-            listener.accept(r, e);
+            try {
+                IOUtils.close(pagesToClose);
+            } catch (Exception ex) {
+                if (e != null) {
+                    ex.addSuppressed(e);
+                }
+                assert false : ex;
+                throw new ElasticsearchException(ex);
+            }
         });
     }
 
@@ -74,7 +78,24 @@ boolean hasEncryptedBytesToFlush() {
 
     @Override
     public void close() {
-        IOUtils.closeWhileHandlingException(currentPage);
-        IOUtils.closeWhileHandlingException(pages);
+        Exception closeException = null;
+        try {
+            IOUtils.close(currentPage);
+        } catch (Exception e) {
+            closeException = e;
+        }
+        currentPage = null;
+        Page p;
+        while ((p = pages.pollFirst()) != null) {
+            try {
+                p.close();
+            } catch (Exception ex) {
+                closeException = ExceptionsHelper.useOrSuppress(closeException, ex);
+            }
+        }
+        if (closeException != null) {
+            assert false : closeException;
+            throw new ElasticsearchException(closeException);
+        }
     }
 }