Optimize slicing ReleasableBytesReference to avoid needless retention of unused pages

server/src/main/java/org/elasticsearch/common/bytes/ReleasableBytesReference.java

@@ -76,11 +76,19 @@ public ReleasableBytesReference retain() {
         return this;
     }
 
+    /**
+     * Same as {@link #slice} except that the slice is not guaranteed to share the same underlying reference count as this instance.
+     * This method is equivalent to calling {@code .slice(from, length).retain()} but might be more efficient through the avoidance of
+     * retaining unnecessary buffers.
+     */
     public ReleasableBytesReference retainedSlice(int from, int length) {
         if (from == 0 && length() == length) {
             return retain();
         }
         final BytesReference slice = delegate.slice(from, length);
+        if (slice instanceof ReleasableBytesReference releasable) {
+            return releasable.retain();
+        }
         refCounted.incRef();
         return new ReleasableBytesReference(slice, refCounted);
     }
@@ -131,10 +139,17 @@ public int length() {
         return delegate.length();
     }
 
+    /**
+     * {@inheritDoc}
+     *
+     * The returned bytes reference will share the reference count of this instance and as such any ref-counting operations on the return
+     * are shared with this instance and vice versa. Using {@link #retainedSlice(int, int)} might be more efficient in situations where the
+     * return of this method is subsequently retained by increasing its ref-count.
+     */
     @Override
-    public BytesReference slice(int from, int length) {
+    public ReleasableBytesReference slice(int from, int length) {
         assert hasReferences();
-        return delegate.slice(from, length);
+        return new ReleasableBytesReference(delegate.slice(from, length), refCounted);
     }
 
     @Override

server/src/main/java/org/elasticsearch/transport/InboundDecoder.java

@@ -121,9 +121,7 @@ public int internalDecode(ReleasableBytesReference reference, CheckedConsumer<Ob
                 bytesConsumedThisDecode += maxBytesToConsume;
                 bytesConsumed += maxBytesToConsume;
                 if (maxBytesToConsume == remainingToConsume) {
-                    try (ReleasableBytesReference retained = reference.retainedSlice(0, maxBytesToConsume)) {
-                        fragmentConsumer.accept(retained);
-                    }
+                    fragmentConsumer.accept(reference.slice(0, maxBytesToConsume));
                 } else {
                     fragmentConsumer.accept(reference);
                 }