Adding RefCounted method implementations to BulkShardRequest

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/SearchCancellationIT.java

@@ -77,21 +77,22 @@ 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++) {
+                    bulkRequestBuilder.add(
+                        prepareIndex("test").setOpType(DocWriteRequest.OpType.CREATE)
+                            .setSource(
+                                "@timestamp",
+                                now + (long) i * numberOfDocsPerRefresh + j,
+                                "val",
+                                (double) j,
+                                "dim",
+                                String.valueOf(j % 100)
+                            )
+                    );
+                }
+                assertNoFailures(bulkRequestBuilder.get());
             }
-            assertNoFailures(bulkRequestBuilder.get());
         }
 
         logger.info("Executing search");

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

@@ -12,6 +12,7 @@
 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;
@@ -502,23 +503,26 @@ 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();
+        }
+        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();
+        }
+        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();
+        }
 
         QueryBuilder queryBuilder = QueryBuilders.rangeQuery("@timestamp").lte("2021-01-01T00:10:00Z");
 

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

@@ -64,15 +64,16 @@ 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);
+                bulkIndexRequest.add(prepareIndex("index").setOpType(DocWriteRequest.OpType.CREATE).setSource(document));
+            }
+
+            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(