Forbid use of Thread.sleep in :server:internalClusterTest

Signed-off-by: Andrew Ross <[email protected]>

Author: andrross

buildSrc/src/main/java/org/opensearch/gradle/precommit/ForbiddenApisPrecommitPlugin.java

@@ -136,6 +136,22 @@ public Void call(Object... names) {
                     return null;
                 }
             });
+            // Add a closure to allow projects to optionally call `forbidSleep()` which will add the signatures
+            // to forbid all usages of `Thread.sleep`
+            ext.set("forbidSleep", new Closure<Void>(t) {
+                @Override
+                public Void call(Object... unused) {
+                    final List<String> signatures = new ArrayList<>();
+                    signatures.addAll(t.getSignatures());
+                    signatures.add(
+                        "java.lang.Thread#sleep(**) @ Fixed sleeps lead to non-deterministic test failures."
+                            + " Poll for whatever condition you're waiting for."
+                            + " Use helpers like `assertBusy` or the awaitility lib."
+                    );
+                    t.setSignatures(signatures);
+                    return null;
+                }
+            });
             // Use of the deprecated security manager APIs are pervasive so set them to warn
             // globally for all projects. Replacements for (most of) these APIs are available
             // so usages can move to the non-deprecated variants to avoid the warnings.

server/build.gradle

@@ -169,6 +169,8 @@ tasks.named("testingConventions").configure {
     }
 }
 
+tasks.named('forbiddenApisInternalClusterTest').configure { forbidSleep() }
+
 // Set to current version by default
 def japicmpCompareTarget = System.getProperty("japicmp.compare.version")
 if (japicmpCompareTarget == null) { /* use latest released version */

server/src/internalClusterTest/java/org/opensearch/action/admin/cluster/shards/TransportCatShardsActionIT.java

@@ -95,27 +95,13 @@ public void testCatShardsWithTimeoutException() throws IOException, AssertionErr
         // Dropping master node to delay in cluster state call.
         internalCluster().stopRandomNode(InternalTestCluster.nameFilter(masterNodes.get(0)));
 
-        CountDownLatch latch = new CountDownLatch(2);
-        new Thread(() -> {
-            try {
-                // Ensures the cancellation timeout expires.
-                Thread.sleep(2000);
-                // Starting master node to proceed in cluster state call.
-                internalCluster().startClusterManagerOnlyNode(
-                    Settings.builder().put("node.name", masterNodes.get(0)).put(clusterManagerDataPathSettings).build()
-                );
-                latch.countDown();
-            } catch (InterruptedException e) {
-                throw new RuntimeException(e);
-            }
-        }).start();
-
+        CountDownLatch latch = new CountDownLatch(1);
         final CatShardsRequest shardsRequest = new CatShardsRequest();
         TimeValue timeoutInterval = timeValueMillis(1000);
         shardsRequest.setCancelAfterTimeInterval(timeoutInterval);
         shardsRequest.clusterManagerNodeTimeout(timeValueMillis(2500));
         shardsRequest.setIndices(Strings.EMPTY_ARRAY);
-        client().execute(CatShardsAction.INSTANCE, shardsRequest, new ActionListener<CatShardsResponse>() {
+        client().execute(CatShardsAction.INSTANCE, shardsRequest, new ActionListener<>() {
             @Override
             public void onResponse(CatShardsResponse catShardsResponse) {
                 // onResponse should not be called.
@@ -132,7 +118,13 @@ public void onFailure(Exception e) {
                 latch.countDown();
             }
         });
+
         latch.await();
+
+        // Restart cluster manager to restore cluster and allow test to cleanup and exit
+        internalCluster().startClusterManagerOnlyNode(
+            Settings.builder().put("node.name", masterNodes.get(0)).put(clusterManagerDataPathSettings).build()
+        );
     }
 
     public void testListShardsWithHiddenIndex() throws Exception {

server/src/internalClusterTest/java/org/opensearch/action/admin/indices/create/RemoteCloneIndexIT.java

@@ -49,6 +49,7 @@
 import org.opensearch.action.admin.indices.stats.IndicesStatsResponse;
 import org.opensearch.cluster.metadata.RepositoryMetadata;
 import org.opensearch.cluster.routing.allocation.decider.EnableAllocationDecider;
+import org.opensearch.common.SuppressForbidden;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.unit.TimeValue;
 import org.opensearch.core.common.unit.ByteSizeValue;
@@ -227,6 +228,7 @@ protected void setLowPriorityUploadRate(String repoName, String value) throws Ex
         createRepository(repoName, rmd.type(), settings);
     }
 
+    @SuppressForbidden(reason = "Waiting longer than timeout")
     public void testCreateCloneIndexFailure() throws ExecutionException, InterruptedException {
         asyncUploadMockFsRepo = false;
         Version version = VersionUtils.randomIndexCompatibleVersion(random());

server/src/internalClusterTest/java/org/opensearch/action/ingest/AsyncIngestProcessorIT.java

@@ -136,13 +136,6 @@ public Map<String, Processor.Factory> getProcessors(Processor.Parameters paramet
                     public void execute(IngestDocument ingestDocument, BiConsumer<IngestDocument, Exception> handler) {
                         threadPool.generic().execute(() -> {
                             String id = (String) ingestDocument.getSourceAndMetadata().get("_id");
-                            if (usually()) {
-                                try {
-                                    Thread.sleep(10);
-                                } catch (InterruptedException e) {
-                                    // ignore
-                                }
-                            }
                             ingestDocument.setFieldValue("foo", "bar-" + id);
                             handler.accept(ingestDocument, null);
                         });

server/src/internalClusterTest/java/org/opensearch/cluster/ClusterInfoServiceIT.java

@@ -44,6 +44,7 @@
 import org.opensearch.cluster.routing.ShardRouting;
 import org.opensearch.cluster.routing.allocation.decider.EnableAllocationDecider;
 import org.opensearch.cluster.service.ClusterService;
+import org.opensearch.common.SuppressForbidden;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.unit.TimeValue;
 import org.opensearch.core.action.ActionListener;
@@ -149,6 +150,7 @@ private void setClusterInfoTimeout(String timeValue) {
         );
     }
 
+    @SuppressForbidden(reason = "Fixed sleeps are prone to flakiness but no documented failures here")
     public void testClusterInfoServiceCollectsInformation() throws InterruptedException {
         Settings.Builder settingsBuilder = Settings.builder()
             .put(ResourceTrackerSettings.GLOBAL_JVM_USAGE_AC_WINDOW_DURATION_SETTING.getKey(), TimeValue.timeValueSeconds(1))
@@ -250,6 +252,7 @@ public void testClusterInfoServiceCollectsFileCacheInformation() {
         }
     }
 
+    @SuppressForbidden(reason = "Fixed sleeps are prone to flakiness but no documented failures here")
     public void testClusterInfoServiceCollectsNodeResourceStatsInformation() throws InterruptedException {
 
         // setting time window as ResourceUsageTracker needs atleast both of these to be ready to start collecting the
@@ -279,6 +282,7 @@ public void testClusterInfoServiceCollectsNodeResourceStatsInformation() throws
         assertEquals(2, nodeResourceUsageStats.size());
     }
 
+    @SuppressForbidden(reason = "Fixed sleeps are prone to flakiness but no documented failures here")
     public void testClusterInfoServiceInformationClearOnError() throws InterruptedException {
         internalCluster().startNodes(
             2,

server/src/internalClusterTest/java/org/opensearch/cluster/coordination/NodeJoinLeftIT.java

@@ -40,6 +40,7 @@
 import org.opensearch.cluster.NodeConnectionsService;
 import org.opensearch.cluster.metadata.IndexMetadata;
 import org.opensearch.cluster.service.ClusterService;
+import org.opensearch.common.SuppressForbidden;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.index.MockEngineFactoryPlugin;
 import org.opensearch.indices.recovery.RecoverySettings;
@@ -75,6 +76,7 @@
  https://github.com/opensearch-project/OpenSearch/pull/15521 for context
  */
 @ClusterScope(scope = Scope.TEST, numDataNodes = 0)
+@SuppressForbidden(reason = "Pending fix: https://github.com/opensearch-project/OpenSearch/issues/18972")
 public class NodeJoinLeftIT extends OpenSearchIntegTestCase {
 
     private TestLogsAppender testLogsAppender;

server/src/internalClusterTest/java/org/opensearch/cluster/coordination/RareClusterStateIT.java

@@ -53,6 +53,7 @@
 import org.opensearch.cluster.routing.allocation.AllocationService;
 import org.opensearch.cluster.routing.allocation.ExistingShardsAllocator;
 import org.opensearch.cluster.service.ClusterService;
+import org.opensearch.common.SuppressForbidden;
 import org.opensearch.common.action.ActionFuture;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.unit.TimeValue;
@@ -219,6 +220,7 @@ public void testDeleteCreateInOneBulk() throws Exception {
         assertHitCount(client().prepareSearch("test").get(), 0);
     }
 
+    @SuppressForbidden(reason = "Best effort sleep, not critical to test logic")
     public void testDelayedMappingPropagationOnPrimary() throws Exception {
         // Here we want to test that things go well if there is a first request
         // that adds mappings but before mappings are propagated to all nodes
@@ -309,6 +311,7 @@ public void testDelayedMappingPropagationOnPrimary() throws Exception {
         assertEquals(1, docIndexResponse.get(10, TimeUnit.SECONDS).getShardInfo().getTotal());
     }
 
+    @SuppressForbidden(reason = "Best effort sleep, not critical to test logic")
     public void testDelayedMappingPropagationOnReplica() throws Exception {
         // This is essentially the same thing as testDelayedMappingPropagationOnPrimary
         // but for replicas

server/src/internalClusterTest/java/org/opensearch/cluster/routing/WeightedRoutingIT.java

@@ -335,15 +335,14 @@ public void testGetWeightedRouting_ClusterManagerNotDiscovered() throws Exceptio
         logger.info("--> network disruption is started");
         networkDisruption.startDisrupting();
 
-        // wait for leader checker to fail
-        Thread.sleep(13000);
-
         // get api to fetch local weighted routing for a node in zone a or b
-        ClusterGetWeightedRoutingResponse weightedRoutingResponse = internalCluster().client(
-            randomFrom(nodes_in_zone_a.get(0), nodes_in_zone_b.get(0))
-        ).admin().cluster().prepareGetWeightedRouting().setAwarenessAttribute("zone").setRequestLocal(true).get();
-        assertEquals(weightedRouting, weightedRoutingResponse.weights());
-        assertFalse(weightedRoutingResponse.getDiscoveredClusterManager());
+        assertBusy(() -> {
+            ClusterGetWeightedRoutingResponse weightedRoutingResponse = internalCluster().client(
+                randomFrom(nodes_in_zone_a.get(0), nodes_in_zone_b.get(0))
+            ).admin().cluster().prepareGetWeightedRouting().setAwarenessAttribute("zone").setRequestLocal(true).get();
+            assertEquals(weightedRouting, weightedRoutingResponse.weights());
+            assertFalse(weightedRoutingResponse.getDiscoveredClusterManager());
+        }, 13, TimeUnit.SECONDS);
         logger.info("--> network disruption is stopped");
         networkDisruption.stopDisrupting();
 

server/src/internalClusterTest/java/org/opensearch/cluster/service/ClusterServiceIT.java

@@ -38,6 +38,7 @@
 import org.opensearch.cluster.ClusterStateUpdateTask;
 import org.opensearch.cluster.node.DiscoveryNode;
 import org.opensearch.common.Nullable;
+import org.opensearch.common.SuppressForbidden;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.unit.TimeValue;
 import org.opensearch.test.OpenSearchIntegTestCase;
@@ -55,6 +56,8 @@
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.hasSize;
+import static org.awaitility.Awaitility.await;
 
 @ClusterScope(scope = Scope.TEST, numDataNodes = 0)
 public class ClusterServiceIT extends OpenSearchIntegTestCase {
@@ -350,6 +353,7 @@ public void testPendingUpdateTask() throws Exception {
         final CountDownLatch invoked1 = new CountDownLatch(1);
         clusterService.submitStateUpdateTask("1", new ClusterStateUpdateTask() {
             @Override
+            @SuppressForbidden(reason = "Sleeping to guarantee a >0 time metric calculation")
             public ClusterState execute(ClusterState currentState) {
                 try {
                     Thread.sleep(50);
@@ -458,10 +462,11 @@ public void onFailure(String source, Exception e) {
                 }
             });
         }
-        Thread.sleep(100);
 
-        pendingClusterTasks = clusterService.getClusterManagerService().pendingTasks();
-        assertThat(pendingClusterTasks.size(), greaterThanOrEqualTo(5));
+        pendingClusterTasks = await().until(
+            () -> clusterService.getClusterManagerService().pendingTasks(),
+            hasSize(greaterThanOrEqualTo(5))
+        );
         controlSources = new HashSet<>(Arrays.asList("1", "2", "3", "4", "5"));
         for (PendingClusterTask task : pendingClusterTasks) {
             controlSources.remove(task.getSource().string());