Fix more search response leaks (#103956)

Some more mechanical fixing of leaked SearchResponse instances.

modules/percolator/src/test/java/org/elasticsearch/percolator/PercolatorQuerySearchTests.java

@@ -9,7 +9,6 @@ package org.elasticsearch.percolator;
 
 import org.apache.lucene.search.Query;
 import org.apache.lucene.search.join.ScoreMode;
-import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.action.support.WriteRequest;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.bytes.BytesReference;
@@ -50,6 +49,7 @@ import static org.elasticsearch.index.query.QueryBuilders.rangeQuery;
 import static org.elasticsearch.index.query.QueryBuilders.scriptQuery;
 import static org.elasticsearch.index.query.QueryBuilders.termQuery;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertHitCount;
+import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertResponse;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertSearchHitsWithoutFailures;
 import static org.elasticsearch.xcontent.XContentFactory.jsonBuilder;
 import static org.hamcrest.Matchers.equalTo;
@@ -407,25 +407,26 @@ public class PercolatorQuerySearchTests extends ESSingleNodeTestCase {
             """);
 
         QueryBuilder query = new PercolateQueryBuilder("my_query", List.of(house1_doc, house2_doc), XContentType.JSON);
-        SearchResponse response = client().prepareSearch("houses").setQuery(query).get();
-        assertEquals(2, response.getHits().getTotalHits().value);
+        assertResponse(client().prepareSearch("houses").setQuery(query), response -> {
+            assertEquals(2, response.getHits().getTotalHits().value);
 
-        SearchHit[] hits = response.getHits().getHits();
-        assertThat(hits[0].getFields().get("_percolator_document_slot").getValues(), equalTo(Arrays.asList(0, 1)));
-        assertThat(
-            hits[0].getFields().get("_percolator_document_slot_0_matched_queries").getValues(),
-            equalTo(Arrays.asList("fireplace_query", "detached_query", "3_bedrooms_query"))
-        );
-        assertThat(
-            hits[0].getFields().get("_percolator_document_slot_1_matched_queries").getValues(),
-            equalTo(Arrays.asList("fireplace_query", "3_bedrooms_query"))
-        );
+            SearchHit[] hits = response.getHits().getHits();
+            assertThat(hits[0].getFields().get("_percolator_document_slot").getValues(), equalTo(Arrays.asList(0, 1)));
+            assertThat(
+                hits[0].getFields().get("_percolator_document_slot_0_matched_queries").getValues(),
+                equalTo(Arrays.asList("fireplace_query", "detached_query", "3_bedrooms_query"))
+            );
+            assertThat(
+                hits[0].getFields().get("_percolator_document_slot_1_matched_queries").getValues(),
+                equalTo(Arrays.asList("fireplace_query", "3_bedrooms_query"))
+            );
 
-        assertThat(hits[1].getFields().get("_percolator_document_slot").getValues(), equalTo(Arrays.asList(0)));
-        assertThat(
-            hits[1].getFields().get("_percolator_document_slot_0_matched_queries").getValues(),
-            equalTo(Arrays.asList("swimming_pool_query", "3_bedrooms_query"))
-        );
+            assertThat(hits[1].getFields().get("_percolator_document_slot").getValues(), equalTo(Arrays.asList(0)));
+            assertThat(
+                hits[1].getFields().get("_percolator_document_slot_0_matched_queries").getValues(),
+                equalTo(Arrays.asList("swimming_pool_query", "3_bedrooms_query"))
+            );
+        });
     }
 
 }

modules/reindex/src/test/java/org/elasticsearch/reindex/AsyncBulkByScrollActionTests.java

@@ -589,30 +589,33 @@ public class AsyncBulkByScrollActionTests extends ESTestCase {
             null,
             SearchResponse.Clusters.EMPTY
         );
+        try {
+            client.lastSearch.get().listener.onResponse(searchResponse);
 
-        client.lastSearch.get().listener.onResponse(searchResponse);
+            assertEquals(0, capturedDelay.get().seconds());
+            capturedCommand.get().run();
 
-        assertEquals(0, capturedDelay.get().seconds());
-        capturedCommand.get().run();
+            // So the next request is going to have to wait an extra 100 seconds or so (base was 10 seconds, so 110ish)
+            assertThat(client.lastScroll.get().request.scroll().keepAlive().seconds(), either(equalTo(110L)).or(equalTo(109L)));
 
-        // So the next request is going to have to wait an extra 100 seconds or so (base was 10 seconds, so 110ish)
-        assertThat(client.lastScroll.get().request.scroll().keepAlive().seconds(), either(equalTo(110L)).or(equalTo(109L)));
+            // Now we can simulate a response and check the delay that we used for the task
+            if (randomBoolean()) {
+                client.lastScroll.get().listener.onResponse(searchResponse);
+                assertEquals(99, capturedDelay.get().seconds());
+            } else {
+                // Let's rethrottle between the starting the scroll and getting the response
+                worker.rethrottle(10f);
+                client.lastScroll.get().listener.onResponse(searchResponse);
+                // The delay uses the new throttle
+                assertEquals(9, capturedDelay.get().seconds());
+            }
 
-        // Now we can simulate a response and check the delay that we used for the task
-        if (randomBoolean()) {
-            client.lastScroll.get().listener.onResponse(searchResponse);
-            assertEquals(99, capturedDelay.get().seconds());
-        } else {
-            // Let's rethrottle between the starting the scroll and getting the response
-            worker.rethrottle(10f);
-            client.lastScroll.get().listener.onResponse(searchResponse);
-            // The delay uses the new throttle
-            assertEquals(9, capturedDelay.get().seconds());
+            // Running the command ought to increment the delay counter on the task.
+            capturedCommand.get().run();
+            assertEquals(capturedDelay.get(), testTask.getStatus().getThrottled());
+        } finally {
+            searchResponse.decRef();
         }
-
-        // Running the command ought to increment the delay counter on the task.
-        capturedCommand.get().run();
-        assertEquals(capturedDelay.get(), testTask.getStatus().getThrottled());
     }
 
     /**

x-pack/plugin/analytics/src/test/java/org/elasticsearch/xpack/analytics/aggregations/metrics/HistogramPercentileAggregationTests.java

@@ -13,7 +13,6 @@ import org.elasticsearch.action.admin.indices.mapping.put.PutMappingRequest;
 import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
 import org.elasticsearch.action.bulk.BulkRequest;
 import org.elasticsearch.action.index.IndexRequest;
-import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.search.aggregations.AggregationBuilders;
 import org.elasticsearch.search.aggregations.metrics.InternalHDRPercentiles;
@@ -211,11 +210,8 @@ public class HistogramPercentileAggregationTests extends ESSingleNodeTestCase {
         }
         client().admin().indices().refresh(new RefreshRequest("raw", "pre_agg")).get();
 
-        SearchResponse response = client().prepareSearch("raw").setTrackTotalHits(true).get();
-        assertEquals(numDocs, response.getHits().getTotalHits().value);
-
-        response = client().prepareSearch("pre_agg").get();
-        assertEquals(numDocs / frq, response.getHits().getTotalHits().value);
+        assertHitCount(client().prepareSearch("raw").setTrackTotalHits(true), numDocs);
+        assertHitCount(client().prepareSearch("pre_agg"), numDocs / frq);
     }
 
     public void testTDigestHistogram() throws Exception {
@@ -228,17 +224,21 @@ public class HistogramPercentileAggregationTests extends ESSingleNodeTestCase {
             .compression(compression)
             .percentiles(10, 25, 50, 75);
 
-        SearchResponse responseRaw = client().prepareSearch("raw").addAggregation(builder).get();
-        SearchResponse responsePreAgg = client().prepareSearch("pre_agg").addAggregation(builder).get();
-        SearchResponse responseBoth = client().prepareSearch("raw", "pre_agg").addAggregation(builder).get();
-
-        InternalTDigestPercentiles percentilesRaw = responseRaw.getAggregations().get("agg");
-        InternalTDigestPercentiles percentilesPreAgg = responsePreAgg.getAggregations().get("agg");
-        InternalTDigestPercentiles percentilesBoth = responseBoth.getAggregations().get("agg");
-        for (int i = 1; i < 100; i++) {
-            assertEquals(percentilesRaw.percentile(i), percentilesPreAgg.percentile(i), 1.0);
-            assertEquals(percentilesRaw.percentile(i), percentilesBoth.percentile(i), 1.0);
-        }
+        assertResponse(
+            client().prepareSearch("raw").addAggregation(builder),
+            responseRaw -> assertResponse(
+                client().prepareSearch("pre_agg").addAggregation(builder),
+                responsePreAgg -> assertResponse(client().prepareSearch("raw", "pre_agg").addAggregation(builder), responseBoth -> {
+                    InternalTDigestPercentiles percentilesRaw = responseRaw.getAggregations().get("agg");
+                    InternalTDigestPercentiles percentilesPreAgg = responsePreAgg.getAggregations().get("agg");
+                    InternalTDigestPercentiles percentilesBoth = responseBoth.getAggregations().get("agg");
+                    for (int i = 1; i < 100; i++) {
+                        assertEquals(percentilesRaw.percentile(i), percentilesPreAgg.percentile(i), 1.0);
+                        assertEquals(percentilesRaw.percentile(i), percentilesBoth.percentile(i), 1.0);
+                    }
+                })
+            )
+        );
     }
 
     public void testBoxplotHistogram() throws Exception {
@@ -246,24 +246,28 @@ public class HistogramPercentileAggregationTests extends ESSingleNodeTestCase {
         setupTDigestHistogram(compression);
         BoxplotAggregationBuilder bpBuilder = new BoxplotAggregationBuilder("agg").field("inner.data").compression(compression);
 
-        SearchResponse bpResponseRaw = client().prepareSearch("raw").addAggregation(bpBuilder).get();
-        SearchResponse bpResponsePreAgg = client().prepareSearch("pre_agg").addAggregation(bpBuilder).get();
-        SearchResponse bpResponseBoth = client().prepareSearch("raw", "pre_agg").addAggregation(bpBuilder).get();
-
-        Boxplot bpRaw = bpResponseRaw.getAggregations().get("agg");
-        Boxplot bpPreAgg = bpResponsePreAgg.getAggregations().get("agg");
-        Boxplot bpBoth = bpResponseBoth.getAggregations().get("agg");
-        assertEquals(bpRaw.getMax(), bpPreAgg.getMax(), 0.0);
-        assertEquals(bpRaw.getMax(), bpBoth.getMax(), 0.0);
-        assertEquals(bpRaw.getMin(), bpPreAgg.getMin(), 0.0);
-        assertEquals(bpRaw.getMin(), bpBoth.getMin(), 0.0);
+        assertResponse(
+            client().prepareSearch("raw").addAggregation(bpBuilder),
+            bpResponseRaw -> assertResponse(
+                client().prepareSearch("pre_agg").addAggregation(bpBuilder),
+                bpResponsePreAgg -> assertResponse(client().prepareSearch("raw", "pre_agg").addAggregation(bpBuilder), bpResponseBoth -> {
+                    Boxplot bpRaw = bpResponseRaw.getAggregations().get("agg");
+                    Boxplot bpPreAgg = bpResponsePreAgg.getAggregations().get("agg");
+                    Boxplot bpBoth = bpResponseBoth.getAggregations().get("agg");
+                    assertEquals(bpRaw.getMax(), bpPreAgg.getMax(), 0.0);
+                    assertEquals(bpRaw.getMax(), bpBoth.getMax(), 0.0);
+                    assertEquals(bpRaw.getMin(), bpPreAgg.getMin(), 0.0);
+                    assertEquals(bpRaw.getMin(), bpBoth.getMin(), 0.0);
 
-        assertEquals(bpRaw.getQ1(), bpPreAgg.getQ1(), 1.0);
-        assertEquals(bpRaw.getQ1(), bpBoth.getQ1(), 1.0);
-        assertEquals(bpRaw.getQ2(), bpPreAgg.getQ2(), 1.0);
-        assertEquals(bpRaw.getQ2(), bpBoth.getQ2(), 1.0);
-        assertEquals(bpRaw.getQ3(), bpPreAgg.getQ3(), 1.0);
-        assertEquals(bpRaw.getQ3(), bpBoth.getQ3(), 1.0);
+                    assertEquals(bpRaw.getQ1(), bpPreAgg.getQ1(), 1.0);
+                    assertEquals(bpRaw.getQ1(), bpBoth.getQ1(), 1.0);
+                    assertEquals(bpRaw.getQ2(), bpPreAgg.getQ2(), 1.0);
+                    assertEquals(bpRaw.getQ2(), bpBoth.getQ2(), 1.0);
+                    assertEquals(bpRaw.getQ3(), bpPreAgg.getQ3(), 1.0);
+                    assertEquals(bpRaw.getQ3(), bpBoth.getQ3(), 1.0);
+                })
+            )
+        );
     }
 
     @Override

x-pack/plugin/async-search/src/internalClusterTest/java/org/elasticsearch/xpack/search/AsyncSearchActionIT.java

@@ -109,38 +109,43 @@ public class AsyncSearchActionIT extends AsyncSearchIntegTestCase {
             .aggregation(AggregationBuilders.max("max").field("metric"));
         try (SearchResponseIterator it = assertBlockingIterator(indexName, numShards, source, numFailures, step)) {
             AsyncSearchResponse response = it.next();
-            while (it.hasNext()) {
-                response = it.next();
-                assertNotNull(response.getSearchResponse());
-                if (response.getSearchResponse().getSuccessfulShards() > 0) {
+            try {
+                while (it.hasNext()) {
+                    response.decRef();
+                    response = it.next();
+                    assertNotNull(response.getSearchResponse());
+                    if (response.getSearchResponse().getSuccessfulShards() > 0) {
+                        assertNotNull(response.getSearchResponse().getAggregations());
+                        assertNotNull(response.getSearchResponse().getAggregations().get("max"));
+                        assertNotNull(response.getSearchResponse().getAggregations().get("min"));
+                        Max max = response.getSearchResponse().getAggregations().get("max");
+                        Min min = response.getSearchResponse().getAggregations().get("min");
+                        assertThat((float) min.value(), greaterThanOrEqualTo(minMetric));
+                        assertThat((float) max.value(), lessThanOrEqualTo(maxMetric));
+                    }
+                }
+                if (numFailures == numShards) {
+                    assertNotNull(response.getFailure());
+                } else {
+                    assertNotNull(response.getSearchResponse());
                     assertNotNull(response.getSearchResponse().getAggregations());
                     assertNotNull(response.getSearchResponse().getAggregations().get("max"));
                     assertNotNull(response.getSearchResponse().getAggregations().get("min"));
                     Max max = response.getSearchResponse().getAggregations().get("max");
                     Min min = response.getSearchResponse().getAggregations().get("min");
-                    assertThat((float) min.value(), greaterThanOrEqualTo(minMetric));
-                    assertThat((float) max.value(), lessThanOrEqualTo(maxMetric));
-                }
-            }
-            if (numFailures == numShards) {
-                assertNotNull(response.getFailure());
-            } else {
-                assertNotNull(response.getSearchResponse());
-                assertNotNull(response.getSearchResponse().getAggregations());
-                assertNotNull(response.getSearchResponse().getAggregations().get("max"));
-                assertNotNull(response.getSearchResponse().getAggregations().get("min"));
-                Max max = response.getSearchResponse().getAggregations().get("max");
-                Min min = response.getSearchResponse().getAggregations().get("min");
-                if (numFailures == 0) {
-                    assertThat((float) min.value(), equalTo(minMetric));
-                    assertThat((float) max.value(), equalTo(maxMetric));
-                } else {
-                    assertThat((float) min.value(), greaterThanOrEqualTo(minMetric));
-                    assertThat((float) max.value(), lessThanOrEqualTo(maxMetric));
+                    if (numFailures == 0) {
+                        assertThat((float) min.value(), equalTo(minMetric));
+                        assertThat((float) max.value(), equalTo(maxMetric));
+                    } else {
+                        assertThat((float) min.value(), greaterThanOrEqualTo(minMetric));
+                        assertThat((float) max.value(), lessThanOrEqualTo(maxMetric));
+                    }
                 }
+                deleteAsyncSearch(response.getId());
+                ensureTaskRemoval(response.getId());
+            } finally {
+                response.decRef();
             }
-            deleteAsyncSearch(response.getId());
-            ensureTaskRemoval(response.getId());
         }
     }
 
@@ -152,10 +157,27 @@ public class AsyncSearchActionIT extends AsyncSearchIntegTestCase {
         );
         try (SearchResponseIterator it = assertBlockingIterator(indexName, numShards, source, numFailures, step)) {
             AsyncSearchResponse response = it.next();
-            while (it.hasNext()) {
-                response = it.next();
-                assertNotNull(response.getSearchResponse());
-                if (response.getSearchResponse().getSuccessfulShards() > 0) {
+            try {
+                while (it.hasNext()) {
+                    response.decRef();
+                    response = it.next();
+                    assertNotNull(response.getSearchResponse());
+                    if (response.getSearchResponse().getSuccessfulShards() > 0) {
+                        assertNotNull(response.getSearchResponse().getAggregations());
+                        assertNotNull(response.getSearchResponse().getAggregations().get("terms"));
+                        StringTerms terms = response.getSearchResponse().getAggregations().get("terms");
+                        assertThat(terms.getBuckets().size(), greaterThanOrEqualTo(0));
+                        assertThat(terms.getBuckets().size(), lessThanOrEqualTo(numKeywords));
+                        for (InternalTerms.Bucket<?> bucket : terms.getBuckets()) {
+                            long count = keywordFreqs.getOrDefault(bucket.getKeyAsString(), new AtomicInteger(0)).get();
+                            assertThat(bucket.getDocCount(), lessThanOrEqualTo(count));
+                        }
+                    }
+                }
+                if (numFailures == numShards) {
+                    assertNotNull(response.getFailure());
+                } else {
+                    assertNotNull(response.getSearchResponse());
                     assertNotNull(response.getSearchResponse().getAggregations());
                     assertNotNull(response.getSearchResponse().getAggregations().get("terms"));
                     StringTerms terms = response.getSearchResponse().getAggregations().get("terms");
@@ -163,58 +185,55 @@ public class AsyncSearchActionIT extends AsyncSearchIntegTestCase {
                     assertThat(terms.getBuckets().size(), lessThanOrEqualTo(numKeywords));
                     for (InternalTerms.Bucket<?> bucket : terms.getBuckets()) {
                         long count = keywordFreqs.getOrDefault(bucket.getKeyAsString(), new AtomicInteger(0)).get();
-                        assertThat(bucket.getDocCount(), lessThanOrEqualTo(count));
+                        if (numFailures > 0) {
+                            assertThat(bucket.getDocCount(), lessThanOrEqualTo(count));
+                        } else {
+                            assertThat(bucket.getDocCount(), equalTo(count));
+                        }
                     }
                 }
+                deleteAsyncSearch(response.getId());
+                ensureTaskRemoval(response.getId());
+            } finally {
+                response.decRef();
             }
-            if (numFailures == numShards) {
-                assertNotNull(response.getFailure());
-            } else {
-                assertNotNull(response.getSearchResponse());
-                assertNotNull(response.getSearchResponse().getAggregations());
-                assertNotNull(response.getSearchResponse().getAggregations().get("terms"));
-                StringTerms terms = response.getSearchResponse().getAggregations().get("terms");
-                assertThat(terms.getBuckets().size(), greaterThanOrEqualTo(0));
-                assertThat(terms.getBuckets().size(), lessThanOrEqualTo(numKeywords));
-                for (InternalTerms.Bucket<?> bucket : terms.getBuckets()) {
-                    long count = keywordFreqs.getOrDefault(bucket.getKeyAsString(), new AtomicInteger(0)).get();
-                    if (numFailures > 0) {
-                        assertThat(bucket.getDocCount(), lessThanOrEqualTo(count));
-                    } else {
-                        assertThat(bucket.getDocCount(), equalTo(count));
-                    }
-                }
-            }
-            deleteAsyncSearch(response.getId());
-            ensureTaskRemoval(response.getId());
         }
     }
 
     public void testRestartAfterCompletion() throws Exception {
-        final AsyncSearchResponse initial;
+        final String initialId;
         try (SearchResponseIterator it = assertBlockingIterator(indexName, numShards, new SearchSourceBuilder(), 0, 2)) {
-            initial = it.next();
+            var initial = it.next();
+            try {
+                initialId = initial.getId();
+            } finally {
+                initial.decRef();
+            }
             while (it.hasNext()) {
-                it.next();
+                it.next().decRef();
             }
         }
-        ensureTaskCompletion(initial.getId());
-        restartTaskNode(initial.getId(), indexName);
+        ensureTaskCompletion(initialId);
+        restartTaskNode(initialId, indexName);
 
-        AsyncSearchResponse response = getAsyncSearch(initial.getId());
-        assertNotNull(response.getSearchResponse());
-        assertFalse(response.isRunning());
-        assertFalse(response.isPartial());
+        AsyncSearchResponse response = getAsyncSearch(initialId);
+        try {
+            assertNotNull(response.getSearchResponse());
+            assertFalse(response.isRunning());
+            assertFalse(response.isPartial());
 
-        AsyncStatusResponse statusResponse = getAsyncStatus(initial.getId());
-        assertFalse(statusResponse.isRunning());
-        assertFalse(statusResponse.isPartial());
-        assertEquals(numShards, statusResponse.getTotalShards());
-        assertEquals(numShards, statusResponse.getSuccessfulShards());
-        assertEquals(RestStatus.OK, statusResponse.getCompletionStatus());
+            AsyncStatusResponse statusResponse = getAsyncStatus(initialId);
+            assertFalse(statusResponse.isRunning());
+            assertFalse(statusResponse.isPartial());
+            assertEquals(numShards, statusResponse.getTotalShards());
+            assertEquals(numShards, statusResponse.getSuccessfulShards());
+            assertEquals(RestStatus.OK, statusResponse.getCompletionStatus());
 
-        deleteAsyncSearch(response.getId());
-        ensureTaskRemoval(response.getId());
+            deleteAsyncSearch(response.getId());
+            ensureTaskRemoval(response.getId());
+        } finally {
+            response.decRef();
+        }
     }
 
     public void testDeleteCancelRunningTask() throws Exception {
@@ -223,6 +242,7 @@ public class AsyncSearchActionIT extends AsyncSearchIntegTestCase {
             SearchResponseIterator it = assertBlockingIterator(indexName, numShards, new SearchSourceBuilder(), randomBoolean() ? 1 : 0, 2)
         ) {
             initial = it.next();
+            initial.decRef();
             deleteAsyncSearch(initial.getId());
             it.close();
             ensureTaskCompletion(initial.getId());
@@ -235,6 +255,7 @@ public class AsyncSearchActionIT extends AsyncSearchIntegTestCase {
             SearchResponseIterator it = assertBlockingIterator(indexName, numShards, new SearchSourceBuilder(), randomBoolean() ? 1 : 0, 2)
         ) {
             AsyncSearchResponse response = it.next();
+            response.decRef();
             deleteAsyncSearch(response.getId());
             it.close();
             ensureTaskCompletion(response.getId());
@@ -243,19 +264,28 @@ public class AsyncSearchActionIT extends AsyncSearchIntegTestCase {
     }
 
     public void testCleanupOnFailure() throws Exception {
-        final AsyncSearchResponse initial;
+        final String initialId;
         try (SearchResponseIterator it = assertBlockingIterator(indexName, numShards, new SearchSourceBuilder(), numShards, 2)) {
-            initial = it.next();
+            var resp = it.next();
+            try {
+                initialId = resp.getId();
+            } finally {
+                resp.decRef();
+            }
+        }
+        ensureTaskCompletion(initialId);
+        AsyncSearchResponse response = getAsyncSearch(initialId);
+        try {
+            assertFalse(response.isRunning());
+            assertNotNull(response.getFailure());
+            assertTrue(response.isPartial());
+            assertThat(response.getSearchResponse().getTotalShards(), equalTo(numShards));
+            assertThat(response.getSearchResponse().getShardFailures().length, equalTo(numShards));
+        } finally {
+            response.decRef();
         }
-        ensureTaskCompletion(initial.getId());
-        AsyncSearchResponse response = getAsyncSearch(initial.getId());
-        assertFalse(response.isRunning());
-        assertNotNull(response.getFailure());
-        assertTrue(response.isPartial());
-        assertThat(response.getSearchResponse().getTotalShards(), equalTo(numShards));
-        assertThat(response.getSearchResponse().getShardFailures().length, equalTo(numShards));
-
-        AsyncStatusResponse statusResponse = getAsyncStatus(initial.getId());
+
+        AsyncStatusResponse statusResponse = getAsyncStatus(initialId);
         assertFalse(statusResponse.isRunning());
         assertTrue(statusResponse.isPartial());
         assertEquals(numShards, statusResponse.getTotalShards());
@@ -263,8 +293,8 @@ public class AsyncSearchActionIT extends AsyncSearchIntegTestCase {
         assertEquals(numShards, statusResponse.getFailedShards());
         assertThat(statusResponse.getCompletionStatus().getStatus(), greaterThanOrEqualTo(400));
 
-        deleteAsyncSearch(initial.getId());
-        ensureTaskRemoval(initial.getId());
+        deleteAsyncSearch(initialId);
+        ensureTaskRemoval(initialId);
     }
 
     public void testInvalidId() throws Exception {
@@ -272,13 +302,18 @@ public class AsyncSearchActionIT extends AsyncSearchIntegTestCase {
             SearchResponseIterator it = assertBlockingIterator(indexName, numShards, new SearchSourceBuilder(), randomBoolean() ? 1 : 0, 2)
         ) {
             AsyncSearchResponse response = it.next();
-            ExecutionException exc = expectThrows(ExecutionException.class, () -> getAsyncSearch("invalid"));
-            assertThat(exc.getCause(), instanceOf(IllegalArgumentException.class));
-            assertThat(exc.getMessage(), containsString("invalid id"));
-            while (it.hasNext()) {
-                response = it.next();
+            try {
+                ExecutionException exc = expectThrows(ExecutionException.class, () -> getAsyncSearch("invalid"));
+                assertThat(exc.getCause(), instanceOf(IllegalArgumentException.class));
+                assertThat(exc.getMessage(), containsString("invalid id"));
+                while (it.hasNext()) {
+                    response.decRef();
+                    response = it.next();
+                }
+                assertFalse(response.isRunning());
+            } finally {
+                response.decRef();
             }
-            assertFalse(response.isRunning());
         }
 
         ExecutionException exc = expectThrows(ExecutionException.class, () -> getAsyncStatus("invalid"));
@@ -289,49 +324,75 @@ public class AsyncSearchActionIT extends AsyncSearchIntegTestCase {
     public void testNoIndex() throws Exception {
         SubmitAsyncSearchRequest request = new SubmitAsyncSearchRequest("invalid-*");
         request.setWaitForCompletionTimeout(TimeValue.timeValueMillis(1));
-        AsyncSearchResponse response = submitAsyncSearch(request);
-        assertNotNull(response.getSearchResponse());
-        assertFalse(response.isRunning());
-        assertThat(response.getSearchResponse().getTotalShards(), equalTo(0));
+        {
+            final AsyncSearchResponse response = submitAsyncSearch(request);
+            try {
+                assertNotNull(response.getSearchResponse());
+                assertFalse(response.isRunning());
+                assertThat(response.getSearchResponse().getTotalShards(), equalTo(0));
+            } finally {
+                response.decRef();
+            }
+        }
 
         request = new SubmitAsyncSearchRequest("invalid");
         request.setWaitForCompletionTimeout(TimeValue.timeValueMillis(1));
-        response = submitAsyncSearch(request);
-        assertNull(response.getSearchResponse());
-        assertNotNull(response.getFailure());
-        assertFalse(response.isRunning());
-        Exception exc = response.getFailure();
-        assertThat(exc.getMessage(), containsString("error while executing search"));
-        assertThat(exc.getCause().getMessage(), containsString("no such index"));
+        {
+            final var response = submitAsyncSearch(request);
+            try {
+                assertNull(response.getSearchResponse());
+                assertNotNull(response.getFailure());
+                assertFalse(response.isRunning());
+                Exception exc = response.getFailure();
+                assertThat(exc.getMessage(), containsString("error while executing search"));
+                assertThat(exc.getCause().getMessage(), containsString("no such index"));
+            } finally {
+                response.decRef();
+            }
+        }
     }
 
     public void testCancellation() throws Exception {
         SubmitAsyncSearchRequest request = new SubmitAsyncSearchRequest(indexName);
         request.getSearchRequest().source(new SearchSourceBuilder().aggregation(new CancellingAggregationBuilder("test", randomLong())));
         request.setWaitForCompletionTimeout(TimeValue.timeValueMillis(1));
-        AsyncSearchResponse response = submitAsyncSearch(request);
-        assertNotNull(response.getSearchResponse());
-        assertTrue(response.isRunning());
-        assertThat(response.getSearchResponse().getTotalShards(), equalTo(numShards));
-        assertThat(response.getSearchResponse().getSuccessfulShards(), equalTo(0));
-        assertThat(response.getSearchResponse().getFailedShards(), equalTo(0));
-
-        response = getAsyncSearch(response.getId());
-        assertNotNull(response.getSearchResponse());
-        assertTrue(response.isRunning());
-        assertThat(response.getSearchResponse().getTotalShards(), equalTo(numShards));
-        assertThat(response.getSearchResponse().getSuccessfulShards(), equalTo(0));
-        assertThat(response.getSearchResponse().getFailedShards(), equalTo(0));
-
-        AsyncStatusResponse statusResponse = getAsyncStatus(response.getId());
-        assertTrue(statusResponse.isRunning());
-        assertEquals(numShards, statusResponse.getTotalShards());
-        assertEquals(0, statusResponse.getSuccessfulShards());
-        assertEquals(0, statusResponse.getSkippedShards());
-        assertEquals(0, statusResponse.getFailedShards());
+        final String responseId;
+        {
+            final AsyncSearchResponse response = submitAsyncSearch(request);
+            try {
+                responseId = response.getId();
+                assertNotNull(response.getSearchResponse());
+                assertTrue(response.isRunning());
+                assertThat(response.getSearchResponse().getTotalShards(), equalTo(numShards));
+                assertThat(response.getSearchResponse().getSuccessfulShards(), equalTo(0));
+                assertThat(response.getSearchResponse().getFailedShards(), equalTo(0));
+            } finally {
+                response.decRef();
+            }
+        }
 
-        deleteAsyncSearch(response.getId());
-        ensureTaskRemoval(response.getId());
+        {
+            final var response = getAsyncSearch(responseId);
+            try {
+                assertNotNull(response.getSearchResponse());
+                assertTrue(response.isRunning());
+                assertThat(response.getSearchResponse().getTotalShards(), equalTo(numShards));
+                assertThat(response.getSearchResponse().getSuccessfulShards(), equalTo(0));
+                assertThat(response.getSearchResponse().getFailedShards(), equalTo(0));
+
+                AsyncStatusResponse statusResponse = getAsyncStatus(response.getId());
+                assertTrue(statusResponse.isRunning());
+                assertEquals(numShards, statusResponse.getTotalShards());
+                assertEquals(0, statusResponse.getSuccessfulShards());
+                assertEquals(0, statusResponse.getSkippedShards());
+                assertEquals(0, statusResponse.getFailedShards());
+
+                deleteAsyncSearch(response.getId());
+                ensureTaskRemoval(response.getId());
+            } finally {
+                response.decRef();
+            }
+        }
     }
 
     public void testUpdateRunningKeepAlive() throws Exception {
@@ -339,45 +400,70 @@ public class AsyncSearchActionIT extends AsyncSearchIntegTestCase {
         request.getSearchRequest().source(new SearchSourceBuilder().aggregation(new CancellingAggregationBuilder("test", randomLong())));
         long now = System.currentTimeMillis();
         request.setWaitForCompletionTimeout(TimeValue.timeValueMillis(1));
-        AsyncSearchResponse response = submitAsyncSearch(request);
-        assertNotNull(response.getSearchResponse());
-        assertTrue(response.isRunning());
-        assertThat(response.getSearchResponse().getTotalShards(), equalTo(numShards));
-        assertThat(response.getSearchResponse().getSuccessfulShards(), equalTo(0));
-        assertThat(response.getSearchResponse().getFailedShards(), equalTo(0));
-        assertThat(response.getExpirationTime(), greaterThan(now));
-        long expirationTime = response.getExpirationTime();
-
-        response = getAsyncSearch(response.getId());
-        assertNotNull(response.getSearchResponse());
-        assertTrue(response.isRunning());
-        assertThat(response.getSearchResponse().getTotalShards(), equalTo(numShards));
-        assertThat(response.getSearchResponse().getSuccessfulShards(), equalTo(0));
-        assertThat(response.getSearchResponse().getFailedShards(), equalTo(0));
+        final long expirationTime;
+        final String responseId;
+        {
+            final AsyncSearchResponse response = submitAsyncSearch(request);
+            try {
+                responseId = response.getId();
+                assertNotNull(response.getSearchResponse());
+                assertTrue(response.isRunning());
+                assertThat(response.getSearchResponse().getTotalShards(), equalTo(numShards));
+                assertThat(response.getSearchResponse().getSuccessfulShards(), equalTo(0));
+                assertThat(response.getSearchResponse().getFailedShards(), equalTo(0));
+                assertThat(response.getExpirationTime(), greaterThan(now));
+                expirationTime = response.getExpirationTime();
+            } finally {
+                response.decRef();
+            }
+        }
 
-        response = getAsyncSearch(response.getId(), TimeValue.timeValueDays(10));
-        assertThat(response.getExpirationTime(), greaterThan(expirationTime));
+        final String responseId2;
+        {
+            final AsyncSearchResponse response = getAsyncSearch(responseId);
+            try {
+                responseId2 = response.getId();
+                assertNotNull(response.getSearchResponse());
+                assertTrue(response.isRunning());
+                assertThat(response.getSearchResponse().getTotalShards(), equalTo(numShards));
+                assertThat(response.getSearchResponse().getSuccessfulShards(), equalTo(0));
+                assertThat(response.getSearchResponse().getFailedShards(), equalTo(0));
+            } finally {
+                response.decRef();
+            }
+        }
 
-        assertTrue(response.isRunning());
-        assertThat(response.getSearchResponse().getTotalShards(), equalTo(numShards));
-        assertThat(response.getSearchResponse().getSuccessfulShards(), equalTo(0));
-        assertThat(response.getSearchResponse().getFailedShards(), equalTo(0));
+        final AsyncSearchResponse response = getAsyncSearch(responseId2, TimeValue.timeValueDays(10));
+        try {
+            assertThat(response.getExpirationTime(), greaterThan(expirationTime));
+
+            assertTrue(response.isRunning());
+            assertThat(response.getSearchResponse().getTotalShards(), equalTo(numShards));
+            assertThat(response.getSearchResponse().getSuccessfulShards(), equalTo(0));
+            assertThat(response.getSearchResponse().getFailedShards(), equalTo(0));
+
+            AsyncStatusResponse statusResponse = getAsyncStatus(response.getId());
+            assertTrue(statusResponse.isRunning());
+            assertTrue(statusResponse.isPartial());
+            assertThat(statusResponse.getExpirationTime(), greaterThan(expirationTime));
+            assertThat(statusResponse.getStartTime(), lessThan(statusResponse.getExpirationTime()));
+            assertEquals(numShards, statusResponse.getTotalShards());
+            assertEquals(0, statusResponse.getSuccessfulShards());
+            assertEquals(0, statusResponse.getFailedShards());
+            assertEquals(0, statusResponse.getSkippedShards());
+            assertEquals(null, statusResponse.getCompletionStatus());
+        } finally {
+            response.decRef();
+        }
 
-        AsyncStatusResponse statusResponse = getAsyncStatus(response.getId());
-        assertTrue(statusResponse.isRunning());
-        assertTrue(statusResponse.isPartial());
-        assertThat(statusResponse.getExpirationTime(), greaterThan(expirationTime));
-        assertThat(statusResponse.getStartTime(), lessThan(statusResponse.getExpirationTime()));
-        assertEquals(numShards, statusResponse.getTotalShards());
-        assertEquals(0, statusResponse.getSuccessfulShards());
-        assertEquals(0, statusResponse.getFailedShards());
-        assertEquals(0, statusResponse.getSkippedShards());
-        assertEquals(null, statusResponse.getCompletionStatus());
-
-        response = getAsyncSearch(response.getId(), TimeValue.timeValueMillis(1));
-        assertThat(response.getExpirationTime(), lessThan(expirationTime));
-        ensureTaskNotRunning(response.getId());
-        ensureTaskRemoval(response.getId());
+        final AsyncSearchResponse response2 = getAsyncSearch(response.getId(), TimeValue.timeValueMillis(1));
+        try {
+            assertThat(response2.getExpirationTime(), lessThan(expirationTime));
+            ensureTaskNotRunning(response2.getId());
+            ensureTaskRemoval(response2.getId());
+        } finally {
+            response2.decRef();
+        }
     }
 
     public void testUpdateStoreKeepAlive() throws Exception {
@@ -386,34 +472,50 @@ public class AsyncSearchActionIT extends AsyncSearchIntegTestCase {
         request.setWaitForCompletionTimeout(TimeValue.timeValueMinutes(10));
         request.setKeepOnCompletion(true);
         AsyncSearchResponse response = submitAsyncSearch(request);
-        assertNotNull(response.getSearchResponse());
-        assertFalse(response.isRunning());
-        assertThat(response.getSearchResponse().getTotalShards(), equalTo(numShards));
-        assertThat(response.getSearchResponse().getSuccessfulShards(), equalTo(numShards));
-        assertThat(response.getSearchResponse().getFailedShards(), equalTo(0));
-        assertThat(response.getExpirationTime(), greaterThan(now));
+        try {
+            assertNotNull(response.getSearchResponse());
+            assertFalse(response.isRunning());
+            assertThat(response.getSearchResponse().getTotalShards(), equalTo(numShards));
+            assertThat(response.getSearchResponse().getSuccessfulShards(), equalTo(numShards));
+            assertThat(response.getSearchResponse().getFailedShards(), equalTo(0));
+            assertThat(response.getExpirationTime(), greaterThan(now));
 
+        } finally {
+            response.decRef();
+        }
         final String searchId = response.getId();
         long expirationTime = response.getExpirationTime();
 
         response = getAsyncSearch(searchId);
-        assertNotNull(response.getSearchResponse());
-        assertFalse(response.isRunning());
-        assertThat(response.getSearchResponse().getTotalShards(), equalTo(numShards));
-        assertThat(response.getSearchResponse().getSuccessfulShards(), equalTo(numShards));
-        assertThat(response.getSearchResponse().getFailedShards(), equalTo(0));
+        try {
+            assertNotNull(response.getSearchResponse());
+            assertFalse(response.isRunning());
+            assertThat(response.getSearchResponse().getTotalShards(), equalTo(numShards));
+            assertThat(response.getSearchResponse().getSuccessfulShards(), equalTo(numShards));
+            assertThat(response.getSearchResponse().getFailedShards(), equalTo(0));
+        } finally {
+            response.decRef();
+        }
 
         response = getAsyncSearch(searchId, TimeValue.timeValueDays(10));
-        assertThat(response.getExpirationTime(), greaterThan(expirationTime));
+        try {
+            assertThat(response.getExpirationTime(), greaterThan(expirationTime));
 
-        assertFalse(response.isRunning());
-        assertThat(response.getSearchResponse().getTotalShards(), equalTo(numShards));
-        assertThat(response.getSearchResponse().getSuccessfulShards(), equalTo(numShards));
-        assertThat(response.getSearchResponse().getFailedShards(), equalTo(0));
+            assertFalse(response.isRunning());
+            assertThat(response.getSearchResponse().getTotalShards(), equalTo(numShards));
+            assertThat(response.getSearchResponse().getSuccessfulShards(), equalTo(numShards));
+            assertThat(response.getSearchResponse().getFailedShards(), equalTo(0));
+        } finally {
+            response.decRef();
+        }
 
         try {
             AsyncSearchResponse finalResponse = getAsyncSearch(searchId, TimeValue.timeValueMillis(1));
-            assertThat(finalResponse.getExpirationTime(), lessThan(expirationTime));
+            try {
+                assertThat(finalResponse.getExpirationTime(), lessThan(expirationTime));
+            } finally {
+                finalResponse.decRef();
+            }
         } catch (ExecutionException e) {
             // The 'get async search' method first updates the expiration time, then gets the response. So the
             // maintenance service might remove the document right after it's updated, which means the get request
@@ -433,18 +535,24 @@ public class AsyncSearchActionIT extends AsyncSearchIntegTestCase {
         request.setKeepOnCompletion(true);
         long now = System.currentTimeMillis();
 
-        AsyncSearchResponse response = submitAsyncSearch(request);
-        assertNotNull(response.getSearchResponse());
-        assertFalse(response.isRunning());
-        assertThat(response.getSearchResponse().getTotalShards(), equalTo(numShards));
-        assertThat(response.getSearchResponse().getSuccessfulShards(), equalTo(numShards));
-        assertThat(response.getSearchResponse().getFailedShards(), equalTo(0));
-        assertThat(response.getExpirationTime(), greaterThan(now));
+        final String responseId;
+        final AsyncSearchResponse response = submitAsyncSearch(request);
+        try {
+            assertNotNull(response.getSearchResponse());
+            assertFalse(response.isRunning());
+            assertThat(response.getSearchResponse().getTotalShards(), equalTo(numShards));
+            assertThat(response.getSearchResponse().getSuccessfulShards(), equalTo(numShards));
+            assertThat(response.getSearchResponse().getFailedShards(), equalTo(0));
+            assertThat(response.getExpirationTime(), greaterThan(now));
+            responseId = response.getId();
+        } finally {
+            response.decRef();
+        }
 
         // remove the async search index
         indicesAdmin().prepareDelete(XPackPlugin.ASYNC_RESULTS_INDEX).get();
 
-        Exception exc = expectThrows(Exception.class, () -> getAsyncSearch(response.getId()));
+        Exception exc = expectThrows(Exception.class, () -> getAsyncSearch(responseId));
         Throwable cause = exc instanceof ExecutionException
             ? ExceptionsHelper.unwrapCause(exc.getCause())
             : ExceptionsHelper.unwrapCause(exc);
@@ -453,16 +561,20 @@ public class AsyncSearchActionIT extends AsyncSearchIntegTestCase {
         SubmitAsyncSearchRequest newReq = new SubmitAsyncSearchRequest(indexName);
         newReq.getSearchRequest().source(new SearchSourceBuilder().aggregation(new CancellingAggregationBuilder("test", randomLong())));
         newReq.setWaitForCompletionTimeout(TimeValue.timeValueMillis(1)).setKeepAlive(TimeValue.timeValueSeconds(1));
-        AsyncSearchResponse newResp = submitAsyncSearch(newReq);
-        assertNotNull(newResp.getSearchResponse());
-        assertTrue(newResp.isRunning());
-        assertThat(newResp.getSearchResponse().getTotalShards(), equalTo(numShards));
-        assertThat(newResp.getSearchResponse().getSuccessfulShards(), equalTo(0));
-        assertThat(newResp.getSearchResponse().getFailedShards(), equalTo(0));
-
-        // check garbage collection
-        ensureTaskNotRunning(newResp.getId());
-        ensureTaskRemoval(newResp.getId());
+        final AsyncSearchResponse newResp = submitAsyncSearch(newReq);
+        try {
+            assertNotNull(newResp.getSearchResponse());
+            assertTrue(newResp.isRunning());
+            assertThat(newResp.getSearchResponse().getTotalShards(), equalTo(numShards));
+            assertThat(newResp.getSearchResponse().getSuccessfulShards(), equalTo(0));
+            assertThat(newResp.getSearchResponse().getFailedShards(), equalTo(0));
+
+            // check garbage collection
+            ensureTaskNotRunning(newResp.getId());
+            ensureTaskRemoval(newResp.getId());
+        } finally {
+            newResp.decRef();
+        }
     }
 
     public void testSearchPhaseFailure() throws Exception {
@@ -473,11 +585,15 @@ public class AsyncSearchActionIT extends AsyncSearchIntegTestCase {
         request.getSearchRequest()
             .source(new SearchSourceBuilder().query(new ThrowingQueryBuilder(randomLong(), new AlreadyClosedException("boom"), 0)));
         AsyncSearchResponse response = submitAsyncSearch(request);
-        assertFalse(response.isRunning());
-        assertTrue(response.isPartial());
-        assertThat(response.status(), equalTo(RestStatus.SERVICE_UNAVAILABLE));
-        assertNotNull(response.getFailure());
-        ensureTaskNotRunning(response.getId());
+        try {
+            assertFalse(response.isRunning());
+            assertTrue(response.isPartial());
+            assertThat(response.status(), equalTo(RestStatus.SERVICE_UNAVAILABLE));
+            assertNotNull(response.getFailure());
+            ensureTaskNotRunning(response.getId());
+        } finally {
+            response.decRef();
+        }
     }
 
     public void testSearchPhaseFailureLeak() throws Exception {
@@ -494,11 +610,15 @@ public class AsyncSearchActionIT extends AsyncSearchIntegTestCase {
         request.getSearchRequest().source().aggregation(terms("f").field("f").size(between(1, 10)));
 
         AsyncSearchResponse response = submitAsyncSearch(request);
-        assertFalse(response.isRunning());
-        assertTrue(response.isPartial());
-        assertThat(response.status(), equalTo(RestStatus.SERVICE_UNAVAILABLE));
-        assertNotNull(response.getFailure());
-        ensureTaskNotRunning(response.getId());
+        try {
+            assertFalse(response.isRunning());
+            assertTrue(response.isPartial());
+            assertThat(response.status(), equalTo(RestStatus.SERVICE_UNAVAILABLE));
+            assertNotNull(response.getFailure());
+            ensureTaskNotRunning(response.getId());
+        } finally {
+            response.decRef();
+        }
     }
 
     public void testMaxResponseSize() {
@@ -538,9 +658,13 @@ public class AsyncSearchActionIT extends AsyncSearchIntegTestCase {
         }), indexName);
 
         AsyncSearchResponse response = submitAsyncSearch(request);
-        assertFalse(response.isRunning());
-        Exception failure = response.getFailure();
-        assertThat(failure.getMessage(), containsString("error while executing search"));
-        assertThat(failure.getCause().getMessage(), containsString("the 'search.check_ccs_compatibility' setting is enabled"));
+        try {
+            assertFalse(response.isRunning());
+            Exception failure = response.getFailure();
+            assertThat(failure.getMessage(), containsString("error while executing search"));
+            assertThat(failure.getCause().getMessage(), containsString("the 'search.check_ccs_compatibility' setting is enabled"));
+        } finally {
+            response.decRef();
+        }
     }
 }