Fix potential circuit breaker leak on InternalGeoGrid (#88273) (#88298)

During reduce time we are creating a LongObjectPagedHashMap that will not get released if there is an error when 
it is getting populated. This commit adds a finally clause so it is always release.

Author: iverase

docs/changelog/88273.yaml

@@ -0,0 +1,6 @@
+pr: 88273
+summary: Fix potential circuit breaker leak on `InternalGeoGrid`
+area: Geo
+type: bug
+issues:
+ - 88261

server/src/main/java/org/elasticsearch/search/aggregations/bucket/geogrid/InternalGeoGrid.java

@@ -11,6 +11,7 @@
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.util.LongObjectPagedHashMap;
+import org.elasticsearch.core.Releasables;
 import org.elasticsearch.search.aggregations.AggregationReduceContext;
 import org.elasticsearch.search.aggregations.InternalAggregation;
 import org.elasticsearch.search.aggregations.InternalAggregations;
@@ -78,30 +79,35 @@ public List<InternalGeoGridBucket> getBuckets() {
     @Override
     public InternalGeoGrid<B> reduce(List<InternalAggregation> aggregations, AggregationReduceContext reduceContext) {
         LongObjectPagedHashMap<List<InternalGeoGridBucket>> buckets = null;
-        for (InternalAggregation aggregation : aggregations) {
-            @SuppressWarnings("unchecked")
-            InternalGeoGrid<B> grid = (InternalGeoGrid<B>) aggregation;
-            if (buckets == null) {
-                buckets = new LongObjectPagedHashMap<>(grid.buckets.size(), reduceContext.bigArrays());
-            }
-            for (Object obj : grid.buckets) {
-                InternalGeoGridBucket bucket = (InternalGeoGridBucket) obj;
-                List<InternalGeoGridBucket> existingBuckets = buckets.get(bucket.hashAsLong());
-                if (existingBuckets == null) {
-                    existingBuckets = new ArrayList<>(aggregations.size());
-                    buckets.put(bucket.hashAsLong(), existingBuckets);
+        final BucketPriorityQueue<InternalGeoGridBucket> ordered;
+        try {
+            for (InternalAggregation aggregation : aggregations) {
+                @SuppressWarnings("unchecked")
+                InternalGeoGrid<B> grid = (InternalGeoGrid<B>) aggregation;
+                if (buckets == null) {
+                    buckets = new LongObjectPagedHashMap<>(grid.buckets.size(), reduceContext.bigArrays());
+                }
+                for (InternalGeoGridBucket bucket : grid.buckets) {
+                    List<InternalGeoGridBucket> existingBuckets = buckets.get(bucket.hashAsLong());
+                    if (existingBuckets == null) {
+                        existingBuckets = new ArrayList<>(aggregations.size());
+                        buckets.put(bucket.hashAsLong(), existingBuckets);
+                    }
+                    existingBuckets.add(bucket);
                 }
-                existingBuckets.add(bucket);
             }
-        }
 
-        final int size = Math.toIntExact(reduceContext.isFinalReduce() == false ? buckets.size() : Math.min(requiredSize, buckets.size()));
-        BucketPriorityQueue<InternalGeoGridBucket> ordered = new BucketPriorityQueue<>(size);
-        for (LongObjectPagedHashMap.Cursor<List<InternalGeoGridBucket>> cursor : buckets) {
-            List<InternalGeoGridBucket> sameCellBuckets = cursor.value;
-            ordered.insertWithOverflow(reduceBucket(sameCellBuckets, reduceContext));
+            final int size = Math.toIntExact(
+                reduceContext.isFinalReduce() == false ? buckets.size() : Math.min(requiredSize, buckets.size())
+            );
+            ordered = new BucketPriorityQueue<>(size);
+            for (LongObjectPagedHashMap.Cursor<List<InternalGeoGridBucket>> cursor : buckets) {
+                List<InternalGeoGridBucket> sameCellBuckets = cursor.value;
+                ordered.insertWithOverflow(reduceBucket(sameCellBuckets, reduceContext));
+            }
+        } finally {
+            Releasables.close(buckets);
         }
-        buckets.close();
         InternalGeoGridBucket[] list = new InternalGeoGridBucket[ordered.size()];
         for (int i = ordered.size() - 1; i >= 0; i--) {
             list[i] = ordered.pop();