Adding RefCounted methods to IndexRequest, UpdateRequest, and ReindexRequest

docs/changelog/102778.yaml

@@ -0,0 +1,5 @@
+pr: 102778
+summary: Adding `RefCounted` method implementations to `BulkRequest`
+area: Ingest Node
+type: enhancement
+issues: []

docs/changelog/103161.yaml

@@ -0,0 +1,5 @@
+pr: 103161
+summary: Adding `RefCounted` method implementations to `BulkShardRequest`
+area: Ingest Node
+type: enhancement
+issues: []

modules/aggregations/src/internalClusterTest/java/org/elasticsearch/aggregations/bucket/AdjacencyMatrixIT.java

@@ -93,6 +93,9 @@ public void setupSuiteScopeCluster() throws Exception {
             );
         }
         indexRandom(true, builders);
+        for (IndexRequestBuilder builder : builders) {
+            builder.request().decRef();
+        }
         ensureSearchable();
     }
 

modules/aggregations/src/internalClusterTest/java/org/elasticsearch/aggregations/bucket/SearchCancellationIT.java

@@ -12,6 +12,7 @@
 import org.elasticsearch.action.ActionFuture;
 import org.elasticsearch.action.DocWriteRequest;
 import org.elasticsearch.action.bulk.BulkRequestBuilder;
+import org.elasticsearch.action.index.IndexRequestBuilder;
 import org.elasticsearch.action.search.SearchPhaseExecutionException;
 import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.action.search.TransportSearchAction;
@@ -77,21 +78,26 @@ public void testCancellationDuringTimeSeriesAggregation() throws Exception {
 
         for (int i = 0; i < numberOfRefreshes; i++) {
             // Make sure we sometimes have a few segments
-            BulkRequestBuilder bulkRequestBuilder = client().prepareBulk().setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
-            for (int j = 0; j < numberOfDocsPerRefresh; j++) {
-                bulkRequestBuilder.add(
-                    prepareIndex("test").setOpType(DocWriteRequest.OpType.CREATE)
-                        .setSource(
-                            "@timestamp",
-                            now + (long) i * numberOfDocsPerRefresh + j,
-                            "val",
-                            (double) j,
-                            "dim",
-                            String.valueOf(j % 100)
-                        )
-                );
+            try (BulkRequestBuilder bulkRequestBuilder = client().prepareBulk().setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE)) {
+                for (int j = 0; j < numberOfDocsPerRefresh; j++) {
+                    IndexRequestBuilder indexRequestBuilder = prepareIndex("test");
+                    try {
+                        indexRequestBuilder.setOpType(DocWriteRequest.OpType.CREATE)
+                            .setSource(
+                                "@timestamp",
+                                now + (long) i * numberOfDocsPerRefresh + j,
+                                "val",
+                                (double) j,
+                                "dim",
+                                String.valueOf(j % 100)
+                            );
+                        bulkRequestBuilder.add(indexRequestBuilder);
+                    } finally {
+                        indexRequestBuilder.request().decRef();
+                    }
+                }
+                assertNoFailures(bulkRequestBuilder.get());
             }
-            assertNoFailures(bulkRequestBuilder.get());
         }
 
         logger.info("Executing search");

modules/aggregations/src/internalClusterTest/java/org/elasticsearch/aggregations/bucket/TimeSeriesAggregationsIT.java

@@ -12,12 +12,14 @@
 import org.elasticsearch.action.DocWriteRequest;
 import org.elasticsearch.action.admin.indices.alias.Alias;
 import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
+import org.elasticsearch.action.bulk.BulkRequestBuilder;
 import org.elasticsearch.action.index.IndexRequestBuilder;
 import org.elasticsearch.action.support.WriteRequest;
 import org.elasticsearch.aggregations.AggregationIntegTestCase;
 import org.elasticsearch.aggregations.bucket.timeseries.InternalTimeSeries;
 import org.elasticsearch.aggregations.bucket.timeseries.TimeSeriesAggregationBuilder;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.core.RefCounted;
 import org.elasticsearch.index.mapper.DateFieldMapper;
 import org.elasticsearch.index.query.QueryBuilder;
 import org.elasticsearch.index.query.QueryBuilders;
@@ -173,6 +175,9 @@ public void setupSuiteScopeCluster() throws Exception {
             docs.add(prepareIndex("index" + findIndex(timestamp)).setOpType(DocWriteRequest.OpType.CREATE).setSource(docSource));
         }
         indexRandom(true, false, docs);
+        for (IndexRequestBuilder doc : docs) {
+            doc.request().decRef();
+        }
     }
 
     public void testStandAloneTimeSeriesAgg() {
@@ -502,23 +507,41 @@ public void testGetHitsFailure() throws Exception {
             ).setMapping("key", "type=keyword,time_series_dimension=true", "val", "type=double")
         );
 
-        client().prepareBulk()
-            .add(prepareIndex("test").setId("2").setSource("key", "bar", "val", 2, "@timestamp", "2021-01-01T00:00:10Z"))
-            .add(prepareIndex("test").setId("1").setSource("key", "bar", "val", 10, "@timestamp", "2021-01-01T00:00:00Z"))
-            .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE)
-            .get();
-        client().prepareBulk()
-            .add(prepareIndex("test").setId("4").setSource("key", "bar", "val", 50, "@timestamp", "2021-01-01T00:00:30Z"))
-            .add(prepareIndex("test").setId("3").setSource("key", "bar", "val", 40, "@timestamp", "2021-01-01T00:00:20Z"))
-            .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE)
-            .get();
-        client().prepareBulk()
-            .add(prepareIndex("test").setId("7").setSource("key", "foo", "val", 20, "@timestamp", "2021-01-01T00:00:00Z"))
-            .add(prepareIndex("test").setId("8").setSource("key", "foo", "val", 30, "@timestamp", "2021-01-01T00:10:00Z"))
-            .add(prepareIndex("test").setId("5").setSource("key", "baz", "val", 20, "@timestamp", "2021-01-01T00:00:00Z"))
-            .add(prepareIndex("test").setId("6").setSource("key", "baz", "val", 30, "@timestamp", "2021-01-01T00:10:00Z"))
-            .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE)
-            .get();
+        try (BulkRequestBuilder bulkRequestBuilder = client().prepareBulk()) {
+            bulkRequestBuilder.add(prepareIndex("test").setId("2").setSource("key", "bar", "val", 2, "@timestamp", "2021-01-01T00:00:10Z"))
+                .add(prepareIndex("test").setId("1").setSource("key", "bar", "val", 10, "@timestamp", "2021-01-01T00:00:00Z"))
+                .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE)
+                .get();
+            for (DocWriteRequest<?> request : bulkRequestBuilder.request().requests()) {
+                if (request instanceof RefCounted refCounted) {
+                    refCounted.decRef();
+                }
+            }
+        }
+        try (BulkRequestBuilder bulkRequestBuilder = client().prepareBulk()) {
+            bulkRequestBuilder.add(prepareIndex("test").setId("4").setSource("key", "bar", "val", 50, "@timestamp", "2021-01-01T00:00:30Z"))
+                .add(prepareIndex("test").setId("3").setSource("key", "bar", "val", 40, "@timestamp", "2021-01-01T00:00:20Z"))
+                .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE)
+                .get();
+            for (DocWriteRequest<?> request : bulkRequestBuilder.request().requests()) {
+                if (request instanceof RefCounted refCounted) {
+                    refCounted.decRef();
+                }
+            }
+        }
+        try (BulkRequestBuilder bulkRequestBuilder = client().prepareBulk()) {
+            bulkRequestBuilder.add(prepareIndex("test").setId("7").setSource("key", "foo", "val", 20, "@timestamp", "2021-01-01T00:00:00Z"))
+                .add(prepareIndex("test").setId("8").setSource("key", "foo", "val", 30, "@timestamp", "2021-01-01T00:10:00Z"))
+                .add(prepareIndex("test").setId("5").setSource("key", "baz", "val", 20, "@timestamp", "2021-01-01T00:00:00Z"))
+                .add(prepareIndex("test").setId("6").setSource("key", "baz", "val", 30, "@timestamp", "2021-01-01T00:10:00Z"))
+                .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE)
+                .get();
+            for (DocWriteRequest<?> request : bulkRequestBuilder.request().requests()) {
+                if (request instanceof RefCounted refCounted) {
+                    refCounted.decRef();
+                }
+            }
+        }
 
         QueryBuilder queryBuilder = QueryBuilders.rangeQuery("@timestamp").lte("2021-01-01T00:10:00Z");
 

modules/aggregations/src/internalClusterTest/java/org/elasticsearch/aggregations/bucket/TimeSeriesNestedAggregationsIT.java

@@ -12,6 +12,7 @@
 import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
 import org.elasticsearch.action.bulk.BulkRequestBuilder;
 import org.elasticsearch.action.bulk.BulkResponse;
+import org.elasticsearch.action.index.IndexRequestBuilder;
 import org.elasticsearch.action.support.WriteRequest;
 import org.elasticsearch.aggregations.AggregationIntegTestCase;
 import org.elasticsearch.aggregations.bucket.timeseries.InternalTimeSeries;
@@ -64,15 +65,21 @@ public void setup() throws Exception {
         assertTrue(prepareTimeSeriesIndex(mapping, startMillis, endMillis, routingDimensions).isAcknowledged());
         logger.info("Dimensions: " + numberOfDimensions + " docs: " + numberOfDocuments + " start: " + startMillis + " end: " + endMillis);
 
-        final BulkRequestBuilder bulkIndexRequest = client().prepareBulk();
-        for (int docId = 0; docId < numberOfDocuments; docId++) {
-            final XContentBuilder document = timeSeriesDocument(FOO_DIM_VALUE, BAR_DIM_VALUE, BAZ_DIM_VALUE, docId, timestamps::next);
-            bulkIndexRequest.add(prepareIndex("index").setOpType(DocWriteRequest.OpType.CREATE).setSource(document));
+        try (BulkRequestBuilder bulkIndexRequest = client().prepareBulk()) {
+            for (int docId = 0; docId < numberOfDocuments; docId++) {
+                final XContentBuilder document = timeSeriesDocument(FOO_DIM_VALUE, BAR_DIM_VALUE, BAZ_DIM_VALUE, docId, timestamps::next);
+                IndexRequestBuilder indexRequestBuilder = prepareIndex("index");
+                try {
+                    bulkIndexRequest.add(indexRequestBuilder.setOpType(DocWriteRequest.OpType.CREATE).setSource(document));
+                } finally {
+                    indexRequestBuilder.request().decRef();
+                }
+            }
+
+            final BulkResponse bulkIndexResponse = bulkIndexRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE).get();
+            assertFalse(bulkIndexResponse.hasFailures());
+            assertEquals(RestStatus.OK.getStatus(), client().admin().indices().prepareFlush("index").get().getStatus().getStatus());
         }
-
-        final BulkResponse bulkIndexResponse = bulkIndexRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE).get();
-        assertFalse(bulkIndexResponse.hasFailures());
-        assertEquals(RestStatus.OK.getStatus(), client().admin().indices().prepareFlush("index").get().getStatus().getStatus());
     }
 
     private static XContentBuilder timeSeriesDocument(

modules/aggregations/src/internalClusterTest/java/org/elasticsearch/aggregations/pipeline/DateDerivativeIT.java

@@ -103,6 +103,9 @@ public void setupSuiteScopeCluster() throws Exception {
             )
         ); // date: Mar 23, dates: Mar 23, Apr 24
         indexRandom(true, builders);
+        for (IndexRequestBuilder builder : builders) {
+            builder.request().decRef();
+        }
         ensureSearchable();
     }
 
@@ -217,6 +220,9 @@ public void testSingleValuedFieldNormalised_timeZone_CET_DstStart() throws Excep
         addNTimes(3, IDX_DST_START, DateFormatters.from(formatter.parse("2012-03-26T01:00:00")), builders);
         addNTimes(4, IDX_DST_START, DateFormatters.from(formatter.parse("2012-03-27T01:00:00")), builders);
         indexRandom(true, builders);
+        for (IndexRequestBuilder builder : builders) {
+            builder.request().decRef();
+        }
         ensureSearchable();
 
         assertNoFailuresAndResponse(
@@ -274,6 +280,9 @@ public void testSingleValuedFieldNormalised_timeZone_CET_DstEnd() throws Excepti
         addNTimes(3, IDX_DST_END, DateFormatters.from(formatter.parse("2012-10-29T01:00:00")), builders);
         addNTimes(4, IDX_DST_END, DateFormatters.from(formatter.parse("2012-10-30T01:00:00")), builders);
         indexRandom(true, builders);
+        for (IndexRequestBuilder builder : builders) {
+            builder.request().decRef();
+        }
         ensureSearchable();
 
         assertNoFailuresAndResponse(
@@ -333,6 +342,9 @@ public void testSingleValuedFieldNormalised_timeZone_AsiaKathmandu() throws Exce
         addNTimes(3, IDX_DST_KATHMANDU, DateFormatters.from(formatter.parse("1986-01-01T01:30:00")), builders);
         addNTimes(4, IDX_DST_KATHMANDU, DateFormatters.from(formatter.parse("1986-01-01T02:30:00")), builders);
         indexRandom(true, builders);
+        for (IndexRequestBuilder builder : builders) {
+            builder.request().decRef();
+        }
         ensureSearchable();
 
         assertNoFailuresAndResponse(

modules/aggregations/src/internalClusterTest/java/org/elasticsearch/aggregations/pipeline/SerialDiffIT.java

@@ -150,6 +150,9 @@ public void setupSuiteScopeCluster() throws Exception {
         }
 
         indexRandom(true, builders);
+        for (IndexRequestBuilder builder : builders) {
+            builder.request().decRef();
+        }
         ensureSearchable();
     }
 

modules/analysis-common/src/internalClusterTest/java/org/elasticsearch/analysis/common/QueryStringWithAnalyzersIT.java

@@ -49,7 +49,7 @@ public void testCustomWordDelimiterQueryString() {
                 .setMapping("field1", "type=text,analyzer=my_analyzer", "field2", "type=text,analyzer=my_analyzer")
         );
 
-        prepareIndex("test").setId("1").setSource("field1", "foo bar baz", "field2", "not needed").get();
+        indexDoc("test", "1", "field1", "foo bar baz", "field2", "not needed");
         refresh();
 
         assertHitCount(

modules/analysis-common/src/internalClusterTest/java/org/elasticsearch/analysis/common/ReloadSynonymAnalyzerIT.java

@@ -70,7 +70,7 @@ private void testSynonymsUpdate(boolean preview) throws FileNotFoundException, I
                 .setMapping("field", "type=text,analyzer=standard,search_analyzer=my_synonym_analyzer")
         );
 
-        prepareIndex("test").setId("1").setSource("field", "foo").get();
+        indexDoc("test", "1", "field", "foo");
         assertNoFailures(indicesAdmin().prepareRefresh("test").get());
 
         assertHitCount(prepareSearch("test").setQuery(QueryBuilders.matchQuery("field", "baz")), 1L);

modules/analysis-common/src/test/java/org/elasticsearch/analysis/common/HighlighterWithAnalyzersTests.java

@@ -8,6 +8,7 @@
 
 package org.elasticsearch.analysis.common;
 
+import org.elasticsearch.action.index.IndexRequestBuilder;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.query.Operator;
@@ -109,7 +110,9 @@ public void testNgramHighlightingWithBrokenPositions() throws IOException {
                         .putList("analysis.analyzer.search_autocomplete.filter", "lowercase", "wordDelimiter")
                 )
         );
-        prepareIndex("test").setId("1").setSource("name", "ARCOTEL Hotels Deutschland").get();
+        IndexRequestBuilder indexRequestBuilder = prepareIndex("test").setId("1").setSource("name", "ARCOTEL Hotels Deutschland");
+        indexRequestBuilder.get();
+        indexRequestBuilder.request().decRef();
         refresh();
         assertResponse(
             prepareSearch("test").setQuery(matchQuery("name.autocomplete", "deut tel").operator(Operator.OR))
@@ -151,16 +154,17 @@ public void testMultiPhraseCutoff() throws IOException {
         );
 
         ensureGreen();
-        prepareIndex("test").setId("1")
+        IndexRequestBuilder indexRequestBuilder = prepareIndex("test").setId("1")
             .setSource(
                 "body",
                 "Test: http://www.facebook.com http://elasticsearch.org "
                     + "http://xing.com http://cnn.com http://quora.com http://twitter.com this is "
                     + "a test for highlighting feature Test: http://www.facebook.com "
                     + "http://elasticsearch.org http://xing.com http://cnn.com http://quora.com "
                     + "http://twitter.com this is a test for highlighting feature"
-            )
-            .get();
+            );
+        indexRequestBuilder.get();
+        indexRequestBuilder.request().decRef();
         refresh();
         assertResponse(
             prepareSearch().setQuery(matchPhraseQuery("body", "Test: http://www.facebook.com "))
@@ -211,7 +215,10 @@ public void testSynonyms() throws IOException {
         );
         ensureGreen();
 
-        prepareIndex("test").setId("0").setSource("field1", "The quick brown fox jumps over the lazy dog").get();
+        IndexRequestBuilder indexRequestBuilder = prepareIndex("test").setId("0")
+            .setSource("field1", "The quick brown fox jumps over the lazy dog");
+        indexRequestBuilder.get();
+        indexRequestBuilder.request().decRef();
         refresh();
         for (String highlighterType : new String[] { "plain", "fvh", "unified" }) {
             logger.info("--> highlighting (type=" + highlighterType + ") and searching on field1");
@@ -248,10 +255,23 @@ public void testPhrasePrefix() throws IOException {
 
         ensureGreen();
 
-        prepareIndex("first_test_index").setId("0")
-            .setSource("field0", "The quick brown fox jumps over the lazy dog", "field1", "The quick brown fox jumps over the lazy dog")
-            .get();
-        prepareIndex("first_test_index").setId("1").setSource("field1", "The quick browse button is a fancy thing, right bro?").get();
+        {
+            IndexRequestBuilder indexRequestBuilder = prepareIndex("first_test_index").setId("0")
+                .setSource(
+                    "field0",
+                    "The quick brown fox jumps over the lazy dog",
+                    "field1",
+                    "The quick brown fox jumps over the lazy dog"
+                );
+            indexRequestBuilder.get();
+            indexRequestBuilder.request().decRef();
+        }
+        {
+            IndexRequestBuilder indexRequestBuilder = prepareIndex("first_test_index").setId("1")
+                .setSource("field1", "The quick browse button is a fancy thing, right bro?");
+            indexRequestBuilder.get();
+            indexRequestBuilder.request().decRef();
+        }
         refresh();
         logger.info("--> highlighting and searching on field0");
 
@@ -324,20 +344,31 @@ public void testPhrasePrefix() throws IOException {
                     "type=text,analyzer=synonym"
                 )
         );
-        prepareIndex("second_test_index").setId("0")
-            .setSource(
-                "type",
-                "type2",
-                "field4",
-                "The quick brown fox jumps over the lazy dog",
-                "field3",
-                "The quick brown fox jumps over the lazy dog"
-            )
-            .get();
-        prepareIndex("second_test_index").setId("1")
-            .setSource("type", "type2", "field4", "The quick browse button is a fancy thing, right bro?")
-            .get();
-        prepareIndex("second_test_index").setId("2").setSource("type", "type2", "field4", "a quick fast blue car").get();
+        {
+            IndexRequestBuilder indexRequestBuilder = prepareIndex("second_test_index").setId("0")
+                .setSource(
+                    "type",
+                    "type2",
+                    "field4",
+                    "The quick brown fox jumps over the lazy dog",
+                    "field3",
+                    "The quick brown fox jumps over the lazy dog"
+                );
+            indexRequestBuilder.get();
+            indexRequestBuilder.request().decRef();
+        }
+        {
+            IndexRequestBuilder indexRequestBuilder = prepareIndex("second_test_index").setId("1")
+                .setSource("type", "type2", "field4", "The quick browse button is a fancy thing, right bro?");
+            indexRequestBuilder.get();
+            indexRequestBuilder.request().decRef();
+        }
+        {
+            IndexRequestBuilder indexRequestBuilder = prepareIndex("second_test_index").setId("2")
+                .setSource("type", "type2", "field4", "a quick fast blue car");
+            indexRequestBuilder.get();
+            indexRequestBuilder.request().decRef();
+        }
         refresh();
 
         assertResponse(