AWS, Core: Use Awaitility instead of Thread.sleep #8804

nk1506 · 2023-10-11T19:02:13Z

nk1506 · 2023-10-12T13:04:03Z

@nastra , Please take a look.

nastra · 2023-10-17T16:32:49Z

api/src/test/java/org/apache/iceberg/TestHelpers.java

@@ -62,6 +70,54 @@ public static long waitUntilAfter(long timestampMillis) {
    return current;
  }

+  /** wait for fixed duration */
+  public static void delayInMilliseconds(String message, long delay, boolean useSameThread) {


I don't think we want to use Awaitility to just wait a certain amount of time for a true condition to happen

nastra · 2023-10-17T16:33:52Z

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

@@ -101,14 +103,7 @@ public void closeableTimer() throws InterruptedException {
  @Test
  public void measureRunnable() {
    Timer timer = new DefaultTimer(TimeUnit.NANOSECONDS);
-    Runnable runnable =


the goal here is actually to measure the duration of the runnable, so this isn't something that we want to replace with Awaitility

nastra · 2023-10-17T16:35:07Z

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

+        "wait for IAM role policy to update.",
+        1000,
+        1001,
+        10001,


I think the parameters are rather confusing. It would be better to use Awaitility directly here

nastra · 2023-10-17T16:35:33Z

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

+        10001,
+        TimeUnit.MILLISECONDS,
+        () ->
+            iam.getRolePolicy(


is that the check required whether IAM is consistent?

nastra · 2023-10-17T16:38:14Z

aws/src/integration/java/org/apache/iceberg/aws/dynamodb/TestDynamoDbLockManager.java

-          } catch (InterruptedException e) {
-            throw new RuntimeException(e);
-          }
+          TestHelpers.delayInMilliseconds(


this goes back to an earlier comment I had on #8715 (comment). We don't just want to replace Thread.sleep() usage blindly with Awaitility by waiting the same amount of time. The important piece is that we'd need to understand what kind of condition the test is trying to eventually reach, which we can then check by using Awaitility (rather than just sleeping X seconds).

I've opened #8853 and #8852 to give an idea how that might look for other places in the code

nastra

this goes back to an earlier comment I had on #8715 (comment). We don't just want to replace Thread.sleep() usage blindly with Awaitility by waiting the same amount of time. The important piece is that we'd need to understand what kind of condition a particular test is trying to eventually reach, which we can then check by using Awaitility (rather than just sleeping X seconds).

I've opened #8853 and #8852 to give an idea how that might look for other places in the code.

That being said, I think the changes in TestHelpers should be reverted

nastra · 2023-10-19T08:00:29Z

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

+                final int currentFilesCount = numCommittedFiles;
+                Awaitility.await()
+                    .pollInterval(Duration.ofMillis(10))
+                    .atMost(Duration.ofMillis(1000))


Suggested change

.atMost(Duration.ofMillis(1000))

.atMost(Duration.ofSeconds(1))

nastra · 2023-10-19T08:01:16Z

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

-    Thread.sleep(IAM_PROPAGATION_DELAY); // sleep to make sure IAM up to date
+  private static void waitForIamConsistency(String roleName, String policyName) {
+    Awaitility.await()
+        .pollDelay(Duration.ofMillis(1000))


Suggested change

.pollDelay(Duration.ofMillis(1000))

.pollDelay(Duration.ofSeconds(1))

nastra · 2023-10-19T08:01:45Z

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

+                            .build())
+                    .roleName()
+                    .equalsIgnoreCase(roleName));
+    ; // wait to make sure IAM up to date


that seeems like a weird formatting. Maybe just put the comment at the top in L363

nastra · 2023-10-19T08:02:45Z

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

+                Awaitility.await()
+                    .pollInterval(Duration.ofMillis(10))
+                    .atMost(Duration.ofMillis(1000))
+                    .until(() -> !(barrier.get() < currentFilesCount * 2));


Suggested change

.until(() -> !(barrier.get() < currentFilesCount * 2));

.until(() -> barrier.get() >= currentFilesCount * 2);

same for the other places

nastra · 2023-10-19T08:03:23Z

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

-  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.")


@amogh-jahagirdar do you know if that would work for the AWS integration tests?

nastra · 2023-10-19T14:54:45Z

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;


changes on this file need to be reverted because we actually want to sleep in those tests

nastra · 2023-10-20T11:31:03Z

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

+                final int currentFilesCount = numCommittedFiles;
+                Awaitility.await()
+                    .pollInterval(Duration.ofMillis(10))
+                    .pollInSameThread()


why do we need this one here but not in the other places? Also this should have an atMost() configuration

Only this test was becoming flaky with Awaitility . I think adding pollInSameThread for all the similar tests is safe. So I have added pollInSameThread and atMost with 5 minutes to all the other similar tests.
wdyt?

@nk1506 I think the question was more around what's the issue if pollInSameThread is not used?

@amogh-jahagirdar , My intent was to use the same test thread here. Since we have multiple runnable tasks here. I don't think it was flaky because of not using pollInSameThread() .
Please share your thoughts about adding pollInSameThread() here is a good idea or not.

then we need to figure out why the test was flaky. Was it occassionally failing? The purpose of using Awaitility is to make tests more robust and less flaky, so we need to figure out why this is flaky

If timeout is less(<10 seconds) it's failing randomly. To safe side, I have added timeout as 5 minutes .atMost(Duration.ofSeconds(300)) .

I'm -1 on having this run 5 minutes. Also I don't think we should be using pollInSameThread(), so please remove all of those. I ran those tests locally with the 10 seconds timeout consistently and they were passing

nastra · 2023-10-25T12:41:22Z

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

+    Awaitility.await("wait for IAM role policy to update.")
+        .pollDelay(Duration.ofSeconds(1))
+        .atMost(Duration.ofSeconds(10))
+        .until(


I think this should be untilAsserted. The inner block would then be assertThat(iam.getRolePolicy(...)).isNotNull()

I think you also might need to specify .ignoreExceptions() as otherwise the API can throw a NoSuchEntityException when the role doesn't exist

I initially planned to add .ignoreExceptions(). But as per there javadoc
Instruct Awaitility to ignore all exceptions that occur during evaluation. Exceptions will be treated as evaluating to false. This is useful in situations where the evaluated conditions may temporarily throw exceptions..
I thought it wont make any difference.
Thanks for recommending this.

amogh-jahagirdar · 2023-10-25T12:46:13Z

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

+          .atMost(Duration.ofSeconds(10))
+          .until(
+              () ->
+                  lakeformation


I believe the same thing @nastra pointed out https://github.com/apache/iceberg/pull/8804/files#r1371698377 applies here as well.

amogh-jahagirdar · 2023-10-25T12:54:41Z

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

+                final int currentFilesCount = numCommittedFiles;
+                Awaitility.await()
+                    .pollInterval(Duration.ofMillis(10))
+                    .pollInSameThread()


@nk1506 I think the question was more around what's the issue if pollInSameThread is not used?

nastra · 2023-10-27T21:40:26Z

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

+    Awaitility.await()
+        .pollDelay(Duration.ofSeconds(1))
+        .atMost(Duration.ofSeconds(10))
+        .until(


this should do the same untilAsserted() check that I mentioned earlier

nastra · 2023-10-27T21:44:53Z

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

+                                  DescribeResourceRequest.builder().resourceArn(arn).build())
+                              .resourceInfo()
+                              .roleArn()
+                              .equalsIgnoreCase(lfRegisterPathRoleArn))


this should do .isEqualToIgnoringCase(lfRegisterPathRoleArn)) rather than isTrue()

nastra · 2023-10-27T21:45:15Z

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

+    Awaitility.await()
+        .pollDelay(Duration.ofSeconds(1))
+        .atMost(Duration.ofSeconds(10))
+        .until(


this should be consistent with other checks

nk1506 · 2023-10-29T04:33:40Z

spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/SparkSQLExecutionHelper.java

@@ -53,7 +53,7 @@ public static String lastExecutedMetricValue(SparkSession spark, String metricNa
    // Refresh metricValues, they will remain null until the execution is complete and metrics are
    // aggregated
    Awaitility.await()
-        .atMost(Duration.ofMillis(500))
+        .atMost(Duration.ofSeconds(10))


It started becoming flaky it seems.
Ref: https://github.com/apache/iceberg/actions/runs/6678213413/job/18149016719

this shouldn't be 10 seconds, but maybe rather 1-3 seconds. The old version of the code was relying on something around 500 millis

nk1506 · 2023-10-29T04:33:50Z

spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/source/SparkSQLExecutionHelper.java

@@ -53,7 +53,7 @@ public static String lastExecutedMetricValue(SparkSession spark, String metricNa
    // Refresh metricValues, they will remain null until the execution is complete and metrics are
    // aggregated
    Awaitility.await()
-        .atMost(Duration.ofMillis(500))
+        .atMost(Duration.ofSeconds(10))


same as above

nk1506 · 2023-10-29T04:34:00Z

spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/SparkSQLExecutionHelper.java

@@ -53,7 +53,7 @@ public static String lastExecutedMetricValue(SparkSession spark, String metricNa
    // Refresh metricValues, they will remain null until the execution is complete and metrics are
    // aggregated
    Awaitility.await()
-        .atMost(Duration.ofMillis(500))
+        .atMost(Duration.ofSeconds(10))


nk1506 · 2023-10-29T04:34:08Z

spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/SparkSQLExecutionHelper.java

@@ -53,7 +53,7 @@ public static String lastExecutedMetricValue(SparkSession spark, String metricNa
    // Refresh metricValues, they will remain null until the execution is complete and metrics are
    // aggregated
    Awaitility.await()
-        .atMost(Duration.ofMillis(500))
+        .atMost(Duration.ofSeconds(10))


nastra · 2023-10-30T11:25:49Z

LGTM, thanks @nk1506

github-actions bot added API core build hive AWS labels Oct 11, 2023

nk1506 mentioned this pull request Oct 11, 2023

Thread.sleep() method is replaced with Awaitility #8725

Closed

nk1506 force-pushed the 7154 branch 5 times, most recently from 010f36b to 7b2c7ac Compare October 12, 2023 09:00

nk1506 marked this pull request as ready for review October 12, 2023 10:52

ajantha-bhat requested a review from nastra October 17, 2023 12:43

nastra reviewed Oct 17, 2023

View reviewed changes

nastra requested changes Oct 17, 2023

View reviewed changes

nk1506 force-pushed the 7154 branch from 7b2c7ac to d9ad859 Compare October 19, 2023 07:24

nastra reviewed Oct 19, 2023

View reviewed changes

nk1506 force-pushed the 7154 branch from d9ad859 to 9bc33bd Compare October 19, 2023 10:26

nk1506 requested a review from nastra October 19, 2023 14:35

nastra reviewed Oct 19, 2023

View reviewed changes

nk1506 force-pushed the 7154 branch from 9bc33bd to e192262 Compare October 19, 2023 15:24

github-actions bot removed the API label Oct 19, 2023

nastra reviewed Oct 20, 2023

View reviewed changes

nk1506 force-pushed the 7154 branch 2 times, most recently from 7577141 to e571385 Compare October 24, 2023 11:13

nk1506 requested a review from nastra October 25, 2023 05:01

nastra reviewed Oct 25, 2023

View reviewed changes

amogh-jahagirdar reviewed Oct 25, 2023

View reviewed changes

nk1506 force-pushed the 7154 branch 2 times, most recently from db71fe1 to 07885ee Compare October 26, 2023 05:53

nk1506 requested a review from nastra October 27, 2023 04:45

nastra reviewed Oct 27, 2023

View reviewed changes

nk1506 force-pushed the 7154 branch 2 times, most recently from 37258a8 to daea635 Compare October 28, 2023 19:00

github-actions bot added the spark label Oct 28, 2023

nk1506 commented Oct 29, 2023

View reviewed changes

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

98294e1

nk1506 force-pushed the 7154 branch from daea635 to 98294e1 Compare October 30, 2023 10:12

nastra changed the title ~~Build: Replace Thread.Sleep() usage with org.Awaitility from Tests.~~ AWS, Core: Use Awaitility instead of Thread.sleep Oct 30, 2023

nastra approved these changes Oct 30, 2023

View reviewed changes

nastra merged commit 7e9e02c into apache:main Oct 30, 2023
45 checks passed

	.atMost(Duration.ofMillis(1000))
	.atMost(Duration.ofSeconds(1))

	.pollDelay(Duration.ofMillis(1000))
	.pollDelay(Duration.ofSeconds(1))

	.until(() -> !(barrier.get() < currentFilesCount * 2));
	.until(() -> barrier.get() >= currentFilesCount * 2);

AWS, Core: Use Awaitility instead of Thread.sleep #8804

AWS, Core: Use Awaitility instead of Thread.sleep #8804

Conversation

nk1506 commented Oct 11, 2023

nk1506 commented Oct 12, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nastra left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nastra commented Oct 30, 2023