Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
shreyanshR7 wants to merge 13 commits into from
Closed

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
7965c31
Thread.sleep() method is replaced with Awaitility
shreyanshR7 Oct 5, 2023
4220218
Update TestIcebergSourceContinuous.java
shreyanshR7 Oct 5, 2023
a9a038c
Update TestStreamingMonitorFunction.java
shreyanshR7 Oct 5, 2023
a5408b7
Update TestIcebergSourceContinuous.java
shreyanshR7 Oct 5, 2023
187f0f5
Update TestStreamingMonitorFunction.java
shreyanshR7 Oct 5, 2023
ece224e
Update TestStreamingMonitorFunction.java
shreyanshR7 Oct 6, 2023
953ccb8
Update TestStreamingMonitorFunction.java
shreyanshR7 Oct 6, 2023
6119e5e
Update TestStreamingMonitorFunction.java
shreyanshR7 Oct 6, 2023
2d5a567
Update TestStreamingMonitorFunction.java
shreyanshR7 Oct 6, 2023
d6eb3fe
Update TestStreamingMonitorFunction.java
shreyanshR7 Oct 6, 2023
53e3f53
Update TestStreamingMonitorFunction.java
shreyanshR7 Oct 6, 2023
1f704ee
Update TestIcebergSourceContinuous.java
shreyanshR7 Oct 6, 2023
67c5978
Update TestIcebergSourceContinuous.java
shreyanshR7 Oct 6, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 6 additions & 5 deletions ....15/flink/src/test/java/org/apache/iceberg/flink/source/TestStreamingMonitorFunction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,7 @@
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.awaitility.Awaitility;

@RunWith(Parameterized.class)
public class TestStreamingMonitorFunction extends TableTestBase {
Expand Down Expand Up @@ -113,7 +114,7 @@ public void testConsumeWithoutStartSnapshotId() throws Exception {

Assert.assertTrue(
"Should have expected elements.", latch.await(WAIT_TIME_MILLIS, TimeUnit.MILLISECONDS));
Thread.sleep(1000L);
Awaitility.await().atLeast(1, TimeUnit.SECONDS).until(() -> true);
Copy link
Contributor

@nastra nastra Oct 6, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this goes back to my earlier comment here. We don't want to blindly just change Thread.sleep usage to Awaitility

Copy link
Author

@shreyanshR7 shreyanshR7 Oct 6, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I am new to open source , will it be ok if i replace this with simply Awaitility.await()
.atMost(WAIT_TIME_MILLIS, TimeUnit.MILLISECONDS)
.pollInterval(100, TimeUnit.MILLISECONDS)
.until(() -> latch.getCount() == 0); @nastra or suggest changes please.

Copy link
Contributor

@nk1506 nk1506 Oct 11, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@shreyanshR7 , I also tried to make a fix for this here . You can still continue on flink module.


// Stop the stream task.
function.close();
Expand Down Expand Up @@ -150,7 +151,7 @@ public void testConsumeFromStartSnapshotId() throws Exception {

Assert.assertTrue(
"Should have expected elements.", latch.await(WAIT_TIME_MILLIS, TimeUnit.MILLISECONDS));
Thread.sleep(1000L);
Awaitility.await().atLeast(1, TimeUnit.SECONDS).until(() -> true);

// Stop the stream task.
function.close();
Expand Down Expand Up @@ -186,7 +187,7 @@ public void testConsumeFromStartTag() throws Exception {

Assert.assertTrue(
"Should have expected elements.", latch.await(WAIT_TIME_MILLIS, TimeUnit.MILLISECONDS));
Thread.sleep(1000L);
Awaitility.await().atLeast(1, TimeUnit.SECONDS).until(() -> true);

// Stop the stream task.
function.close();
Expand Down Expand Up @@ -214,7 +215,7 @@ public void testCheckpointRestore() throws Exception {

Assert.assertTrue(
"Should have expected elements.", latch.await(WAIT_TIME_MILLIS, TimeUnit.MILLISECONDS));
Thread.sleep(1000L);
Awaitility.await().atLeast(1, TimeUnit.SECONDS).until(() -> true);

state = harness.snapshot(1, 1);

Expand All @@ -240,7 +241,7 @@ public void testCheckpointRestore() throws Exception {

Assert.assertTrue(
"Should have expected elements.", latch.await(WAIT_TIME_MILLIS, TimeUnit.MILLISECONDS));
Thread.sleep(1000L);
Awaitility.await().atLeast(1, TimeUnit.SECONDS).until(() -> true);

// Stop the stream task.
newFunc.close();
Expand Down
18 changes: 10 additions & 8 deletions ...1.16/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceContinuous.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,7 @@
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import org.awaitility.Awaitility;

public class TestIcebergSourceContinuous {

Expand Down Expand Up @@ -237,7 +238,7 @@ public void testLatestSnapshot() throws Exception {
createStream(scanContext).executeAndCollect(getClass().getSimpleName())) {
// we want to make sure job is running first so that enumerator can
// start from the latest snapshot before inserting the next batch2 below.
waitUntilJobIsRunning(MINI_CLUSTER_RESOURCE.getClusterClient());
Awaitility.await().pollDelay(10, TimeUnit.MILLISECONDS).until(() -> !getRunningJobs(MINI_CLUSTER_RESOURCE.getClusterClient()).isEmpty());

// inclusive behavior for starting snapshot
List<Row> result1 = waitForResult(iter, 2);
Expand Down Expand Up @@ -325,8 +326,7 @@ public void testSpecificSnapshotTimestamp() throws Exception {
long snapshot0Timestamp = tableResource.table().currentSnapshot().timestampMillis();

// sleep for 2 ms to make sure snapshot1 has a higher timestamp value
Thread.sleep(2);

waitUntil1ms();
// snapshot1
List<Record> batch1 =
RandomGenericData.generate(tableResource.table().schema(), 2, randomSeed.incrementAndGet());
Expand Down Expand Up @@ -400,11 +400,13 @@ public static List<Row> waitForResult(CloseableIterator<Row> iter, int limit) {
}
return results;
}

public static void waitUntilJobIsRunning(ClusterClient<?> client) throws Exception {
while (getRunningJobs(client).isEmpty()) {
Thread.sleep(10);
}
public static long waitUntil1ms() {
long previous = System.currentTimeMillis();
long current = System.currentTimeMillis();
while (current <= previous) {
current = System.currentTimeMillis();
}
return current;
}

public static List<JobID> getRunningJobs(ClusterClient<?> client) throws Exception {
Expand Down
17 changes: 7 additions & 10 deletions ....16/flink/src/test/java/org/apache/iceberg/flink/source/TestStreamingMonitorFunction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,7 @@
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.awaitility.Awaitility;

@RunWith(Parameterized.class)
public class TestStreamingMonitorFunction extends TableTestBase {
Expand Down Expand Up @@ -113,8 +114,7 @@ public void testConsumeWithoutStartSnapshotId() throws Exception {

Assert.assertTrue(
"Should have expected elements.", latch.await(WAIT_TIME_MILLIS, TimeUnit.MILLISECONDS));
Thread.sleep(1000L);

Awaitility.await().atLeast(1, TimeUnit.SECONDS).until(() -> true);
// Stop the stream task.
function.close();

Expand Down Expand Up @@ -150,8 +150,7 @@ public void testConsumeFromStartSnapshotId() throws Exception {

Assert.assertTrue(
"Should have expected elements.", latch.await(WAIT_TIME_MILLIS, TimeUnit.MILLISECONDS));
Thread.sleep(1000L);

Awaitility.await().atLeast(1, TimeUnit.SECONDS).until(() -> true);
// Stop the stream task.
function.close();

Expand Down Expand Up @@ -186,8 +185,7 @@ public void testConsumeFromStartTag() throws Exception {

Assert.assertTrue(
"Should have expected elements.", latch.await(WAIT_TIME_MILLIS, TimeUnit.MILLISECONDS));
Thread.sleep(1000L);

Awaitility.await().atLeast(1, TimeUnit.SECONDS).until(() -> true);
// Stop the stream task.
function.close();

Expand All @@ -214,8 +212,8 @@ public void testCheckpointRestore() throws Exception {

Assert.assertTrue(
"Should have expected elements.", latch.await(WAIT_TIME_MILLIS, TimeUnit.MILLISECONDS));
Thread.sleep(1000L);

Awaitility.await().atLeast(1, TimeUnit.SECONDS).until(() -> true);
state = harness.snapshot(1, 1);

// Stop the stream task.
Expand All @@ -240,8 +238,7 @@ public void testCheckpointRestore() throws Exception {

Assert.assertTrue(
"Should have expected elements.", latch.await(WAIT_TIME_MILLIS, TimeUnit.MILLISECONDS));
Thread.sleep(1000L);

Awaitility.await().atLeast(1, TimeUnit.SECONDS).until(() -> true);
// Stop the stream task.
newFunc.close();

Expand Down
17 changes: 10 additions & 7 deletions ...1.17/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceContinuous.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,7 @@
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import org.awaitility.Awaitility;

public class TestIcebergSourceContinuous {

Expand Down Expand Up @@ -237,7 +238,7 @@ public void testLatestSnapshot() throws Exception {
createStream(scanContext).executeAndCollect(getClass().getSimpleName())) {
// we want to make sure job is running first so that enumerator can
// start from the latest snapshot before inserting the next batch2 below.
waitUntilJobIsRunning(MINI_CLUSTER_RESOURCE.getClusterClient());
Awaitility.await().pollDelay(10, TimeUnit.MILLISECONDS).until(() -> !getRunningJobs(MINI_CLUSTER_RESOURCE.getClusterClient()).isEmpty());

// inclusive behavior for starting snapshot
List<Row> result1 = waitForResult(iter, 2);
Expand Down Expand Up @@ -325,7 +326,7 @@ public void testSpecificSnapshotTimestamp() throws Exception {
long snapshot0Timestamp = tableResource.table().currentSnapshot().timestampMillis();

// sleep for 2 ms to make sure snapshot1 has a higher timestamp value
Thread.sleep(2);
waitUntil1ms();

// snapshot1
List<Record> batch1 =
Expand Down Expand Up @@ -400,11 +401,13 @@ public static List<Row> waitForResult(CloseableIterator<Row> iter, int limit) {
}
return results;
}

public static void waitUntilJobIsRunning(ClusterClient<?> client) throws Exception {
while (getRunningJobs(client).isEmpty()) {
Thread.sleep(10);
}
public static long waitUntil1ms() {
long previous = System.currentTimeMillis();
long current = System.currentTimeMillis();
while (current <= previous) {
current = System.currentTimeMillis();
}
return current;
}

public static List<JobID> getRunningJobs(ClusterClient<?> client) throws Exception {
Expand Down
22 changes: 11 additions & 11 deletions ....17/flink/src/test/java/org/apache/iceberg/flink/source/TestStreamingMonitorFunction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,7 @@
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.awaitility.Awaitility;

@RunWith(Parameterized.class)
public class TestStreamingMonitorFunction extends TableTestBase {
Expand Down Expand Up @@ -113,8 +114,8 @@ public void testConsumeWithoutStartSnapshotId() throws Exception {

Assert.assertTrue(
"Should have expected elements.", latch.await(WAIT_TIME_MILLIS, TimeUnit.MILLISECONDS));
Thread.sleep(1000L);

Awaitility.await().atLeast(1, TimeUnit.SECONDS).until(() -> true);
// Stop the stream task.
function.close();

Expand Down Expand Up @@ -147,11 +148,10 @@ public void testConsumeFromStartSnapshotId() throws Exception {
CountDownLatch latch = new CountDownLatch(1);
TestSourceContext sourceContext = new TestSourceContext(latch);
runSourceFunctionInTask(sourceContext, function);

Assert.assertTrue(
"Should have expected elements.", latch.await(WAIT_TIME_MILLIS, TimeUnit.MILLISECONDS));
Thread.sleep(1000L);

Awaitility.await().atLeast(1, TimeUnit.SECONDS).until(() -> true);
// Stop the stream task.
function.close();

Expand Down Expand Up @@ -186,8 +186,8 @@ public void testConsumeFromStartTag() throws Exception {

Assert.assertTrue(
"Should have expected elements.", latch.await(WAIT_TIME_MILLIS, TimeUnit.MILLISECONDS));
Thread.sleep(1000L);

Awaitility.await().atLeast(1, TimeUnit.SECONDS).until(() -> true);
// Stop the stream task.
function.close();

Expand All @@ -214,8 +214,8 @@ public void testCheckpointRestore() throws Exception {

Assert.assertTrue(
"Should have expected elements.", latch.await(WAIT_TIME_MILLIS, TimeUnit.MILLISECONDS));
Thread.sleep(1000L);

Awaitility.await().atLeast(1, TimeUnit.SECONDS).until(() -> true);
state = harness.snapshot(1, 1);

// Stop the stream task.
Expand All @@ -240,8 +240,8 @@ public void testCheckpointRestore() throws Exception {

Assert.assertTrue(
"Should have expected elements.", latch.await(WAIT_TIME_MILLIS, TimeUnit.MILLISECONDS));
Thread.sleep(1000L);

Awaitility.await().atLeast(1, TimeUnit.SECONDS).until(() -> true);
// Stop the stream task.
newFunc.close();

Expand Down