Remove dangerous ByteBufStreamInput methods

modules/transport-netty4/src/main/java/org/elasticsearch/transport/netty4/ByteBufStreamInput.java

@@ -33,34 +33,34 @@
 class ByteBufStreamInput extends StreamInput {
 
     private final ByteBuf buffer;
-    private final int startIndex;
     private final int endIndex;
 
     ByteBufStreamInput(ByteBuf buffer, int length) {
         if (length > buffer.readableBytes()) {
             throw new IndexOutOfBoundsException();
         }
         this.buffer = buffer;
-        startIndex = buffer.readerIndex();
+        int startIndex = buffer.readerIndex();
         endIndex = startIndex + length;
         buffer.markReaderIndex();
     }
 
     @Override
     public BytesReference readBytesReference(int length) throws IOException {
-        BytesReference ref = Netty4Utils.toBytesReference(buffer.slice(buffer.readerIndex(), length));
-        buffer.skipBytes(length);
-        return ref;
+        // NOTE: It is unsafe to share a reference of the internal structure, so we
+        // use the default implementation which will copy the bytes. It is unsafe because
+        // a netty ByteBuf might be pooled which requires a manual release to prevent
+        // memory leaks.
+        return super.readBytesReference(length);
     }
 
     @Override
     public BytesRef readBytesRef(int length) throws IOException {
-        if (!buffer.hasArray()) {
-            return super.readBytesRef(length);
-        }
-        BytesRef bytesRef = new BytesRef(buffer.array(), buffer.arrayOffset() + buffer.readerIndex(), length);
-        buffer.skipBytes(length);
-        return bytesRef;
+        // NOTE: It is unsafe to share a reference of the internal structure, so we
+        // use the default implementation which will copy the bytes. It is unsafe because
+        // a netty ByteBuf might be pooled which requires a manual release to prevent
+        // memory leaks.
+        return super.readBytesRef(length);
     }
 
     @Override