Fix

Author: Tim-Brooks

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

@@ -27,12 +27,12 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.CyclicBarrier;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.locks.LockSupport;
 
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertNoFailures;
 import static org.elasticsearch.xcontent.XContentFactory.jsonBuilder;
 
+@ESIntegTestCase.ClusterScope(scope = ESIntegTestCase.Scope.SUITE, numDataNodes = 1)
 public class BulkSourceReleaseIT extends ESIntegTestCase {
 
     @Override
@@ -58,7 +58,7 @@ public void testBulkSourceReleaseWhenIngestReplacesSource() throws Exception {
         indexRequest.setPipeline(pipelineId);
 
         CountDownLatch blockLatch = new CountDownLatch(1);
-        blockWritePool(internalCluster().getInstance(ThreadPool.class), blockLatch);
+        blockWritePool(internalCluster().getDataNodeInstance(ThreadPool.class), blockLatch);
 
         PlainActionFuture<BulkResponse> future = new PlainActionFuture<>();
 
@@ -103,14 +103,14 @@ public void testBytesReferencedByTwoSourcesNotReleasedIfOnlyOneIngestPipeline()
         assertTrue(originalBytes.hasReferences());
 
         CountDownLatch blockLatch = new CountDownLatch(1);
-        blockWritePool(internalCluster().getInstance(ThreadPool.class), blockLatch);
+        blockWritePool(internalCluster().getDataNodeInstance(ThreadPool.class), blockLatch);
 
         PlainActionFuture<BulkResponse> future = new PlainActionFuture<>();
         try {
             handler.lastItems(List.of(indexRequest, indexRequestNoIngest), future);
 
             // Pause briefly to allow bytes to theoretically be released after ingest processing
-            LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(50));
+            LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(500));
 
             assertTrue(originalBytes.hasReferences());
         } finally {
@@ -159,7 +159,7 @@ public void testSomeReferencesCanBeReleasedWhileOthersRetained() throws Exceptio
         assertTrue(retainedBytes.hasReferences());
 
         CountDownLatch blockLatch = new CountDownLatch(1);
-        blockWritePool(internalCluster().getInstance(ThreadPool.class), blockLatch);
+        blockWritePool(internalCluster().getDataNodeInstance(ThreadPool.class), blockLatch);
 
         PlainActionFuture<BulkResponse> future = new PlainActionFuture<>();
         try {
@@ -218,29 +218,4 @@ public boolean isForceExecution() {
         }
         safeAwait(startBarrier);
     }
-
-    private static void fillWriteQueue(ThreadPool threadPool) {
-        final var queueSize = Math.toIntExact(threadPool.info(ThreadPool.Names.WRITE).getQueueSize().singles());
-        final var queueFilled = new AtomicBoolean(false);
-        final var queueFillingTask = new AbstractRunnable() {
-            @Override
-            public void onFailure(Exception e) {
-                fail(e);
-            }
-
-            @Override
-            protected void doRun() {
-                assertTrue("thread pool not blocked", queueFilled.get());
-            }
-
-            @Override
-            public boolean isForceExecution() {
-                return true;
-            }
-        };
-        for (int i = 0; i < queueSize; i++) {
-            threadPool.executor(ThreadPool.Names.WRITE).execute(queueFillingTask);
-        }
-        queueFilled.set(true);
-    }
 }