Use JDK's cleaner to always pool when reading source in SearchHit

WIP, we can make this a little more efficient by not doing redundant
wrapping but I'd like a test run first.

Author: original-brownbear

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

@@ -84,6 +84,10 @@ public ReleasableBytesReference retainedSlice(int from, int length) {
         return new ReleasableBytesReference(slice, refCounted);
     }
 
+    public RefCounted getRefCounted() {
+        return refCounted;
+    }
+
     @Override
     public void close() {
         refCounted.decRef();

server/src/main/java/org/elasticsearch/search/SearchHit.java

@@ -14,6 +14,7 @@
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.bytes.ReleasableBytesReference;
 import org.elasticsearch.common.compress.CompressorFactory;
 import org.elasticsearch.common.document.DocumentField;
 import org.elasticsearch.common.io.stream.StreamInput;
@@ -210,9 +211,13 @@ public static SearchHit readFrom(StreamInput in, boolean pooled) throws IOExcept
         final long version = in.readLong();
         final long seqNo = in.readZLong();
         final long primaryTerm = in.readVLong();
-        BytesReference source = pooled ? in.readReleasableBytesReference() : in.readBytesReference();
+        ReleasableBytesReference source = in.readReleasableBytesReference();
         if (source.length() == 0) {
             source = null;
+        } else {
+            if (pooled == false) {
+                source = LeakTracker.releaseEventually(source);
+            }
         }
         Explanation explanation = null;
         if (in.readBoolean()) {

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

@@ -11,6 +11,7 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.common.Randomness;
+import org.elasticsearch.common.bytes.ReleasableBytesReference;
 import org.elasticsearch.common.util.concurrent.ConcurrentCollections;
 import org.elasticsearch.common.util.set.Sets;
 import org.elasticsearch.core.Assertions;
@@ -63,6 +64,12 @@ public static void setContextHint(String hint) {
         contextHint = hint;
     }
 
+    public static ReleasableBytesReference releaseEventually(ReleasableBytesReference referent) {
+        final RefCounted refCounted = referent.getRefCounted();
+        var cleanable = cleaner.register(referent, refCounted::decRef);
+        return new ReleasableBytesReference(referent, cleanable::clean);
+    }
+
     public static Releasable wrap(Releasable releasable) {
         if (Assertions.ENABLED == false) {
             return releasable;