Build: Replace Thread.Sleep() usage with org.Awaitility from Tests.

api/src/test/java/org/apache/iceberg/metrics/TestDefaultTimer.java

@@ -20,6 +20,7 @@
 
 import static java.util.concurrent.Executors.newFixedThreadPool;
 import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.apache.iceberg.TestHelpers.waitUntilAfter;
 
 import java.time.Duration;
 import java.util.List;
@@ -87,12 +88,12 @@ public void multipleStops() {
   }
 
   @Test
-  public void closeableTimer() throws InterruptedException {
+  public void closeableTimer() {
     Timer timer = new DefaultTimer(TimeUnit.NANOSECONDS);
     Assertions.assertThat(timer.count()).isEqualTo(0);
     Assertions.assertThat(timer.totalDuration()).isEqualTo(Duration.ZERO);
-    try (Timer.Timed sample = timer.start()) {
-      Thread.sleep(500L);
+    try (Timer.Timed ignored = timer.start()) {
+      waitUntilAfter(System.currentTimeMillis());
     }
     Assertions.assertThat(timer.count()).isEqualTo(1);
     Assertions.assertThat(timer.totalDuration()).isGreaterThan(Duration.ZERO);
@@ -128,11 +129,7 @@ public void measureCallable() throws Exception {
     Timer timer = new DefaultTimer(TimeUnit.NANOSECONDS);
     Callable<Boolean> callable =
         () -> {
-          try {
-            Thread.sleep(100);
-          } catch (InterruptedException e) {
-            throw new RuntimeException(e);
-          }
+          waitUntilAfter(System.currentTimeMillis());
           return true;
         };
     Assertions.assertThat(timer.count()).isEqualTo(0);
@@ -154,11 +151,7 @@ public void measureSupplier() {
     Timer timer = new DefaultTimer(TimeUnit.NANOSECONDS);
     Supplier<Boolean> supplier =
         () -> {
-          try {
-            Thread.sleep(100);
-          } catch (InterruptedException e) {
-            throw new RuntimeException(e);
-          }
+          waitUntilAfter(System.currentTimeMillis());
           return true;
         };
     Assertions.assertThat(timer.count()).isEqualTo(0);

aws/src/integration/java/org/apache/iceberg/aws/TestAssumeRoleAwsClientFactory.java

@@ -18,6 +18,7 @@
  */
 package org.apache.iceberg.aws;
 
+import java.time.Duration;
 import java.util.Map;
 import java.util.UUID;
 import org.apache.iceberg.aws.glue.GlueCatalog;
@@ -27,6 +28,7 @@
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.assertj.core.api.Assertions;
 import org.assertj.core.api.InstanceOfAssertFactories;
+import org.awaitility.Awaitility;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
@@ -42,6 +44,7 @@
 import software.amazon.awssdk.services.iam.model.CreateRoleResponse;
 import software.amazon.awssdk.services.iam.model.DeleteRolePolicyRequest;
 import software.amazon.awssdk.services.iam.model.DeleteRoleRequest;
+import software.amazon.awssdk.services.iam.model.GetRolePolicyRequest;
 import software.amazon.awssdk.services.iam.model.PutRolePolicyRequest;
 import software.amazon.awssdk.services.s3.model.S3Exception;
 
@@ -100,7 +103,7 @@ public void after() {
   }
 
   @Test
-  public void testAssumeRoleGlueCatalog() throws Exception {
+  public void testAssumeRoleGlueCatalog() {
     String glueArnPrefix = "arn:aws:glue:*:" + AwsIntegTestUtil.testAccountId();
     iam.putRolePolicy(
         PutRolePolicyRequest.builder()
@@ -189,7 +192,18 @@ public void testAssumeRoleS3FileIO() throws Exception {
     Assert.assertFalse("should be able to access file", inputFile.exists());
   }
 
-  private void waitForIamConsistency() throws Exception {
-    Thread.sleep(10000); // sleep to make sure IAM up to date
+  private void waitForIamConsistency() {
+    Awaitility.await("wait for IAM role policy to update.")
+        .pollDelay(Duration.ofSeconds(1))
+        .atMost(Duration.ofSeconds(10))
+        .until(
+            () ->
+                iam.getRolePolicy(
+                        GetRolePolicyRequest.builder()
+                            .roleName(roleName)
+                            .roleName(policyName)
+                            .build())
+                    .roleName()
+                    .equalsIgnoreCase(roleName));
   }
 }

aws/src/integration/java/org/apache/iceberg/aws/glue/TestGlueCatalogLock.java

@@ -18,6 +18,7 @@
  */
 package org.apache.iceberg.aws.glue;
 
+import java.time.Duration;
 import java.util.List;
 import java.util.UUID;
 import java.util.concurrent.ExecutorService;
@@ -38,6 +39,7 @@
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.relocated.com.google.common.util.concurrent.MoreExecutors;
 import org.apache.iceberg.util.Tasks;
+import org.awaitility.Awaitility;
 import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.BeforeClass;
@@ -139,14 +141,11 @@ public void testParallelCommitMultiThreadMultiCommit() {
         .run(
             index -> {
               for (int numCommittedFiles = 0; numCommittedFiles < 10; numCommittedFiles++) {
-                while (barrier.get() < numCommittedFiles * 2) {
-                  try {
-                    Thread.sleep(10);
-                  } catch (InterruptedException e) {
-                    throw new RuntimeException(e);
-                  }
-                }
-
+                final int currentFilesCount = numCommittedFiles;
+                Awaitility.await()
+                    .pollInterval(Duration.ofMillis(10))
+                    .atMost(Duration.ofMillis(1000))
+                    .until(() -> barrier.get() >= currentFilesCount * 2);
                 table.newFastAppend().appendFile(file).commit();
                 barrier.incrementAndGet();
               }

aws/src/integration/java/org/apache/iceberg/aws/lakeformation/LakeFormationTestBase.java

@@ -21,6 +21,7 @@
 import java.io.UnsupportedEncodingException;
 import java.net.URLDecoder;
 import java.nio.charset.StandardCharsets;
+import java.time.Duration;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -37,6 +38,7 @@
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.types.Types;
+import org.awaitility.Awaitility;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.slf4j.Logger;
@@ -67,6 +69,7 @@
 import software.amazon.awssdk.services.lakeformation.model.DataLocationResource;
 import software.amazon.awssdk.services.lakeformation.model.DatabaseResource;
 import software.amazon.awssdk.services.lakeformation.model.DeregisterResourceRequest;
+import software.amazon.awssdk.services.lakeformation.model.DescribeResourceRequest;
 import software.amazon.awssdk.services.lakeformation.model.EntityNotFoundException;
 import software.amazon.awssdk.services.lakeformation.model.GetDataLakeSettingsRequest;
 import software.amazon.awssdk.services.lakeformation.model.GetDataLakeSettingsResponse;
@@ -175,7 +178,7 @@ public static void beforeClass() throws Exception {
         lfRegisterPathRoleIamPolicyName,
         lfRegisterPathRolePolicyDocForIam(lfRegisterPathRoleArn),
         lfRegisterPathRoleName);
-    waitForIamConsistency();
+    waitForIamConsistency(lfRegisterPathRoleName, lfRegisterPathRoleIamPolicyName);
 
     // create lfPrivilegedRole
     response =
@@ -205,7 +208,7 @@ public static void beforeClass() throws Exception {
         lfPrivilegedRolePolicyName,
         lfPrivilegedRolePolicyDoc(),
         lfPrivilegedRoleName);
-    waitForIamConsistency();
+    waitForIamConsistency(lfPrivilegedRoleName, lfPrivilegedRolePolicyName);
 
     // build lf and glue client with lfRegisterPathRole
     lakeformation =
@@ -250,7 +253,6 @@ public static void beforeClass() throws Exception {
     // register S3 test bucket path
     deregisterResource(testBucketPath);
     registerResource(testBucketPath);
-    waitForIamConsistency();
   }
 
   @AfterClass
@@ -357,8 +359,20 @@ String getRandomTableName() {
     return LF_TEST_TABLE_PREFIX + UUID.randomUUID().toString().replace("-", "");
   }
 
-  private static void waitForIamConsistency() throws Exception {
-    Thread.sleep(IAM_PROPAGATION_DELAY); // sleep to make sure IAM up to date
+  private static void waitForIamConsistency(String roleName, String policyName) {
+    // wait to make sure IAM up to date
+    Awaitility.await()
+        .pollDelay(Duration.ofSeconds(1))
+        .atMost(Duration.ofSeconds(10))
+        .until(
+            () ->
+                iam.getRolePolicy(
+                        GetRolePolicyRequest.builder()
+                            .roleName(roleName)
+                            .policyName(policyName)
+                            .build())
+                    .roleName()
+                    .equalsIgnoreCase(roleName));
   }
 
   private static LakeFormationClient buildLakeFormationClient(
@@ -417,7 +431,16 @@ private static void registerResource(String s3Location) {
               .build());
       // when a resource is registered, LF will update SLR with necessary permissions which has a
       // propagation delay
-      waitForIamConsistency();
+      Awaitility.await()
+          .pollDelay(Duration.ofSeconds(1))
+          .atMost(Duration.ofSeconds(10))
+          .until(
+              () ->
+                  lakeformation
+                      .describeResource(DescribeResourceRequest.builder().resourceArn(arn).build())
+                      .resourceInfo()
+                      .roleArn()
+                      .equalsIgnoreCase(lfRegisterPathRoleArn));
     } catch (AlreadyExistsException e) {
       LOG.warn("Resource {} already registered. Error: {}", arn, e.getMessage(), e);
     } catch (Exception e) {

aws/src/integration/java/org/apache/iceberg/aws/lakeformation/TestLakeFormationAwsClientFactory.java

@@ -18,6 +18,7 @@
  */
 package org.apache.iceberg.aws.lakeformation;
 
+import java.time.Duration;
 import java.util.Map;
 import java.util.UUID;
 import org.apache.iceberg.aws.AwsIntegTestUtil;
@@ -26,6 +27,7 @@
 import org.apache.iceberg.aws.glue.GlueCatalog;
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.awaitility.Awaitility;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
@@ -41,6 +43,7 @@
 import software.amazon.awssdk.services.iam.model.CreateRoleResponse;
 import software.amazon.awssdk.services.iam.model.DeleteRolePolicyRequest;
 import software.amazon.awssdk.services.iam.model.DeleteRoleRequest;
+import software.amazon.awssdk.services.iam.model.GetRolePolicyRequest;
 import software.amazon.awssdk.services.iam.model.PutRolePolicyRequest;
 
 public class TestLakeFormationAwsClientFactory {
@@ -128,8 +131,18 @@ public void testLakeFormationEnabledGlueCatalog() throws Exception {
                     + glueArnPrefix
                     + ":userDefinedFunction/allowed_*/*\"]}]}")
             .build());
-    waitForIamConsistency();
-
+    Awaitility.await()
+        .pollDelay(Duration.ofSeconds(1))
+        .atMost(Duration.ofSeconds(10))
+        .until(
+            () ->
+                iam.getRolePolicy(
+                        GetRolePolicyRequest.builder()
+                            .roleName(roleName)
+                            .policyName(policyName)
+                            .build())
+                    .roleName()
+                    .equalsIgnoreCase(roleName));
     GlueCatalog glueCatalog = new GlueCatalog();
     assumeRoleProperties.put("warehouse", "s3://path");
     glueCatalog.initialize("test", assumeRoleProperties);
@@ -162,8 +175,4 @@ public void testLakeFormationEnabledGlueCatalog() throws Exception {
       }
     }
   }
-
-  private void waitForIamConsistency() throws Exception {
-    Thread.sleep(IAM_PROPAGATION_DELAY); // sleep to make sure IAM up to date
-  }
 }

build.gradle

@@ -293,6 +293,7 @@ project(':iceberg-api') {
     compileOnly libs.findbugs.jsr305
     testImplementation libs.avro.avro
     testImplementation libs.esotericsoftware.kryo
+    testImplementation libs.awaitility
   }
 
   tasks.processTestResources.dependsOn rootProject.tasks.buildInfo
@@ -492,6 +493,7 @@ project(':iceberg-aws') {
     testImplementation libs.mockserver.client.java
     testImplementation libs.jaxb.api
     testImplementation project(path: ':iceberg-core', configuration: 'testArtifacts')
+    testImplementation libs.awaitility
   }
 
   sourceSets {
@@ -712,6 +714,7 @@ project(':iceberg-hive-metastore') {
     }
 
     testImplementation project(path: ':iceberg-api', configuration: 'testArtifacts')
+    testImplementation libs.awaitility
   }
 }
 

core/src/test/java/org/apache/iceberg/hadoop/TestHadoopCommits.java

@@ -28,6 +28,7 @@
 
 import java.io.File;
 import java.io.IOException;
+import java.time.Duration;
 import java.util.List;
 import java.util.Set;
 import java.util.UUID;
@@ -54,6 +55,7 @@
 import org.apache.iceberg.types.Types;
 import org.apache.iceberg.util.Tasks;
 import org.assertj.core.api.Assertions;
+import org.awaitility.Awaitility;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
 import org.mockito.Mockito;
@@ -435,13 +437,11 @@ public void testConcurrentFastAppends(@TempDir File dir) throws Exception {
               for (int numCommittedFiles = 0;
                   numCommittedFiles < numberOfCommitedFilesPerThread;
                   numCommittedFiles++) {
-                while (barrier.get() < numCommittedFiles * threadsCount) {
-                  try {
-                    Thread.sleep(10);
-                  } catch (InterruptedException e) {
-                    throw new RuntimeException(e);
-                  }
-                }
+                final int currentFilesCount = numCommittedFiles;
+                Awaitility.await()
+                    .pollInterval(Duration.ofMillis(10))
+                    .pollInSameThread()
+                    .until(() -> barrier.get() >= currentFilesCount * threadsCount);
                 tableWithHighRetries.newFastAppend().appendFile(file).commit();
                 barrier.incrementAndGet();
               }

core/src/test/java/org/apache/iceberg/jdbc/TestJdbcTableConcurrency.java

@@ -26,6 +26,7 @@
 
 import java.io.File;
 import java.io.IOException;
+import java.time.Duration;
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.ExecutorService;
@@ -46,6 +47,7 @@
 import org.apache.iceberg.relocated.com.google.common.util.concurrent.MoreExecutors;
 import org.apache.iceberg.types.Types;
 import org.apache.iceberg.util.Tasks;
+import org.awaitility.Awaitility;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
 
@@ -92,14 +94,11 @@ public synchronized void testConcurrentFastAppends() throws IOException {
         .run(
             index -> {
               for (int numCommittedFiles = 0; numCommittedFiles < 10; numCommittedFiles++) {
-                while (barrier.get() < numCommittedFiles * 2) {
-                  try {
-                    Thread.sleep(10);
-                  } catch (InterruptedException e) {
-                    throw new RuntimeException(e);
-                  }
-                }
-
+                final int currentFilesCount = numCommittedFiles;
+                Awaitility.await()
+                    .pollInterval(Duration.ofMillis(10))
+                    .atMost(Duration.ofMillis(1000))
+                    .until(() -> barrier.get() >= currentFilesCount * 2);
                 icebergTable.newFastAppend().appendFile(file).commit();
                 barrier.incrementAndGet();
               }