Unlink delegate in ReleasableBytesReference once ref count reaches 0 (elastic#127058)

We have some spots where we retain a reference to the `ReleasableBytesReference` instance
well beyond its ref-count reaching `0`. If it itself references Netty buffers or `BigArrays`
that are not pooled (mostly as a result of overflowing the pooled number of bytes for large
messages or under heavy load) then those bytes are not GC-able unless we unlink them here.

Author: original-brownbear

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

@@ -29,7 +29,7 @@ public final class ReleasableBytesReference implements RefCounted, Releasable, B
 
     private static final ReleasableBytesReference EMPTY = new ReleasableBytesReference(BytesArray.EMPTY, RefCounted.ALWAYS_REFERENCED);
 
-    private final BytesReference delegate;
+    private BytesReference delegate;
     private final RefCounted refCounted;
 
     public static ReleasableBytesReference empty() {
@@ -63,20 +63,29 @@ public boolean tryIncRef() {
 
     @Override
     public boolean decRef() {
-        return refCounted.decRef();
+        boolean res = refCounted.decRef();
+        if (res) {
+            delegate = null;
+        }
+        return res;
     }
 
     @Override
     public boolean hasReferences() {
-        return refCounted.hasReferences();
+        boolean hasRef = refCounted.hasReferences();
+        // delegate is nulled out when the ref-count reaches zero but only via a plain store, and also we could be racing with a concurrent
+        // decRef so need to check #refCounted again in case we run into a non-null delegate but saw a reference before
+        assert delegate != null || refCounted.hasReferences() == false;
+        return hasRef;
     }
 
     public ReleasableBytesReference retain() {
-        refCounted.incRef();
+        refCounted.mustIncRef();
         return this;
     }
 
     public ReleasableBytesReference retainedSlice(int from, int length) {
+        assert hasReferences();
         if (from == 0 && length() == length) {
             return retain();
         }
@@ -128,6 +137,7 @@ public int indexOf(byte marker, int from) {
 
     @Override
     public int length() {
+        assert hasReferences();
         return delegate.length();
     }
 
@@ -139,6 +149,7 @@ public BytesReference slice(int from, int length) {
 
     @Override
     public long ramBytesUsed() {
+        assert hasReferences();
         return delegate.ramBytesUsed();
     }
 
@@ -213,6 +224,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
 
     @Override
     public boolean isFragment() {
+        assert hasReferences();
         return delegate.isFragment();
     }
 

x-pack/plugin/security/qa/multi-cluster/src/javaRestTest/java/org/elasticsearch/xpack/remotecluster/RemoteClusterSecurityFcActionAuthorizationIT.java

@@ -80,7 +80,6 @@
 import static org.elasticsearch.xpack.remotecluster.AbstractRemoteClusterSecurityTestCase.performRequestWithAdminUser;
 import static org.hamcrest.Matchers.arrayContaining;
 import static org.hamcrest.Matchers.containsString;
-import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.instanceOf;
@@ -282,7 +281,7 @@ public void testIndicesPrivilegesAreEnforcedForCcrRestoreSessionActions() throws
                 GetCcrRestoreFileChunkAction.REMOTE_TYPE,
                 new GetCcrRestoreFileChunkRequest(response2.getNode(), sessionUUID2, leaderIndex2FileName, 1, shardId2)
             );
-            assertThat(getChunkResponse.getChunk().length(), equalTo(1));
+            assertFalse(getChunkResponse.getChunk().hasReferences());
 
             // Clear restore session fails if index is unauthorized
             final var e4 = expectThrows(