Fix

Author: Tim-Brooks

server/src/internalClusterTest/java/org/elasticsearch/action/bulk/BulkSourceReleaseIT.java

@@ -47,7 +47,8 @@ public void testBulkSourceReleaseWhenIngestReplacesSource() throws Exception {
         String pipelineId = "pipeline_id";
         putPipeline(pipelineId);
 
-        IncrementalBulkService incrementalBulkService = internalCluster().getInstance(IncrementalBulkService.class);
+        // Get the data node to ensure the ingest pipeline can be performed on this node
+        IncrementalBulkService incrementalBulkService = internalCluster().getDataNodeInstance(IncrementalBulkService.class);
 
         ReleasableBytesReference originalBytes = new ReleasableBytesReference(new BytesArray("{\"field\": \"value\"}"), () -> {});
 
@@ -80,7 +81,8 @@ public void testBytesReferencedByTwoSourcesNotReleasedIfOnlyOneIngestPipeline()
         String pipelineId = "pipeline_id";
         putPipeline(pipelineId);
 
-        IncrementalBulkService incrementalBulkService = internalCluster().getInstance(IncrementalBulkService.class);
+        // Get the data node to ensure the ingest pipeline can be performed on this node
+        IncrementalBulkService incrementalBulkService = internalCluster().getDataNodeInstance(IncrementalBulkService.class);
 
         ReleasableBytesReference originalBytes = new ReleasableBytesReference(
             new BytesArray("{\"field\": \"value1\"}{\"field\": \"value2\"}"),
@@ -130,7 +132,8 @@ public void testSomeReferencesCanBeReleasedWhileOthersRetained() throws Exceptio
         String pipelineId = "pipeline_id";
         putPipeline(pipelineId);
 
-        IncrementalBulkService incrementalBulkService = internalCluster().getInstance(IncrementalBulkService.class);
+        // Get the data node to ensure the ingest pipeline can be performed on this node
+        IncrementalBulkService incrementalBulkService = internalCluster().getDataNodeInstance(IncrementalBulkService.class);
 
         ReleasableBytesReference releasedBytes = new ReleasableBytesReference(new BytesArray("{\"field\": \"value1\"}"), () -> {});
         ReleasableBytesReference retainedBytes = new ReleasableBytesReference(
@@ -147,15 +150,14 @@ public void testSomeReferencesCanBeReleasedWhileOthersRetained() throws Exceptio
 
         IndexRequest indexRequest2 = new IndexRequest();
         indexRequest2.index(index);
-        indexRequest2.sourceContext().source(retainedBytes.retainedSlice(0, splitPoint), XContentType.JSON);
+        indexRequest2.sourceContext().source(retainedBytes.slice(0, splitPoint), XContentType.JSON);
         indexRequest2.setPipeline(pipelineId);
 
         IndexRequest indexRequestNoIngest = new IndexRequest();
         indexRequestNoIngest.index(index);
         indexRequestNoIngest.sourceContext()
             .source(retainedBytes.retainedSlice(splitPoint, retainedBytes.length() - splitPoint), XContentType.JSON);
 
-        retainedBytes.decRef();
         assertTrue(retainedBytes.hasReferences());
 
         CountDownLatch blockLatch = new CountDownLatch(1);