Release DocumentField instances in SearchHit.deallocate (elastic#124925) (elastic#125071)

These can consume non-trivial heap if transport messages start to queue
up. Let's release them asap.

Author: original-brownbear

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

@@ -520,13 +520,15 @@ public DocumentField removeDocumentField(String field) {
      * @return a map of metadata fields for this hit
      */
     public Map<String, DocumentField> getMetadataFields() {
+        assert hasReferences();
         return Collections.unmodifiableMap(metaFields);
     }
 
     /**
      * @return a map of non-metadata fields requested for this hit
      */
     public Map<String, DocumentField> getDocumentFields() {
+        assert hasReferences();
         return Collections.unmodifiableMap(documentFields);
     }
 
@@ -535,6 +537,7 @@ public Map<String, DocumentField> getDocumentFields() {
      * were required to be loaded. Includes both document and metadata fields.
      */
     public Map<String, DocumentField> getFields() {
+        assert hasReferences();
         if (metaFields.size() > 0 || documentFields.size() > 0) {
             final Map<String, DocumentField> fields = new HashMap<>();
             fields.putAll(metaFields);
@@ -556,6 +559,7 @@ public boolean hasLookupFields() {
      * Resolve the lookup fields with the given results and merge them as regular fetch fields.
      */
     public void resolveLookupFields(Map<LookupField, List<Object>> lookupResults) {
+        assert hasReferences();
         if (lookupResults.isEmpty()) {
             return;
         }
@@ -585,6 +589,7 @@ public void resolveLookupFields(Map<LookupField, List<Object>> lookupResults) {
      * A map of highlighted fields.
      */
     public Map<String, HighlightField> getHighlightFields() {
+        assert hasReferences();
         return highlightFields == null ? emptyMap() : highlightFields;
     }
 
@@ -724,6 +729,17 @@ private void deallocate() {
             r.decRef();
         }
         SearchHit.this.source = null;
+        clearIfMutable(documentFields);
+        clearIfMutable(metaFields);
+        this.highlightFields = null;
+    }
+
+    private static void clearIfMutable(Map<String, DocumentField> fields) {
+        // check that we're dealing with a HashMap, instances read from the wire that are empty be of an immutable type
+        assert fields instanceof HashMap<?, ?> || fields.isEmpty() : fields;
+        if (fields instanceof HashMap<?, ?> hm) {
+            hm.clear();
+        }
     }
 
     @Override
@@ -756,12 +772,16 @@ public SearchHit asUnpooled() {
             innerHits == null
                 ? null
                 : innerHits.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().asUnpooled())),
-            documentFields,
-            metaFields,
+            cloneIfHashMap(documentFields),
+            cloneIfHashMap(metaFields),
             ALWAYS_REFERENCED
         );
     }
 
+    private Map<String, DocumentField> cloneIfHashMap(Map<String, DocumentField> map) {
+        return map instanceof HashMap<String, DocumentField> hashMap ? new HashMap<>(hashMap) : map;
+    }
+
     public boolean isPooled() {
         return refCounted != ALWAYS_REFERENCED;
     }