Skip to content

Commit 36f7539

Browse files
blakeli0blakeli0
committed
refactor(test): migrate Thread.sleep wait loops to Awaitility in GAX tests
1 parent f31da65 commit 36f7539

3 files changed

Lines changed: 93 additions & 86 deletions

File tree

‎sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/batching/BatcherImplTest.java‎

Lines changed: 52 additions & 51 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@
3333
import static com.google.api.gax.rpc.testing.FakeBatchableApi.callLabeledIntSquarer;
3434
import static com.google.common.truth.Truth.assertThat;
3535
import static com.google.common.truth.Truth.assertWithMessage;
36+
import static org.awaitility.Awaitility.await;
3637
import static org.junit.jupiter.api.Assertions.assertThrows;
3738
import static org.mockito.ArgumentMatchers.any;
3839
import static org.mockito.ArgumentMatchers.eq;
@@ -599,15 +600,14 @@ void testPushCurrentBatchRunnable() throws Exception {
599600
// Batcher present inside runnable should be GCed after following loop.
600601
batcher.close();
601602
batcher = null;
602-
for (int retry = 0; retry < 3; retry++) {
603-
System.gc();
604-
System.runFinalization();
605-
isExecutorCancelled = pushBatchRunnable.isCancelled();
606-
if (isExecutorCancelled) {
607-
break;
608-
}
609-
Thread.sleep(DELAY_TIME * (1L << retry));
610-
}
603+
await()
604+
.atMost(Duration.ofSeconds(5))
605+
.until(
606+
() -> {
607+
System.gc();
608+
System.runFinalization();
609+
return pushBatchRunnable.isCancelled();
610+
});
611611
// ScheduledFuture should be isCancelled now.
612612
assertThat(pushBatchRunnable.isCancelled()).isTrue();
613613
}
@@ -735,15 +735,14 @@ public void splitResponse(
735735
void testUnclosedBatchersAreLogged() throws Exception {
736736
final long DELAY_TIME = 30L;
737737
int actualRemaining = 0;
738-
for (int retry = 0; retry < 3; retry++) {
739-
System.gc();
740-
System.runFinalization();
741-
actualRemaining = BatcherReference.cleanQueue();
742-
if (actualRemaining == 0) {
743-
break;
744-
}
745-
Thread.sleep(DELAY_TIME * (1L << retry));
746-
}
738+
await()
739+
.atMost(Duration.ofSeconds(5))
740+
.until(
741+
() -> {
742+
System.gc();
743+
System.runFinalization();
744+
return BatcherReference.cleanQueue() == 0;
745+
});
747746
assertThat(actualRemaining).isAtMost(0);
748747
underTest = createDefaultBatcherImpl(batchingSettings, null);
749748
Batcher<Integer, Integer> extraBatcher = createDefaultBatcherImpl(batchingSettings, null);
@@ -771,20 +770,16 @@ public boolean isLoggable(LogRecord record) {
771770

772771
underTest = null;
773772
// That *should* have been the last reference. Try to reclaim it.
774-
boolean success = false;
775-
for (int retry = 0; retry < 3; retry++) {
776-
System.gc();
777-
System.runFinalization();
778-
int orphans = BatcherReference.cleanQueue();
779-
if (orphans == 1) {
780-
success = true;
781-
break;
782-
}
783-
// Validates that there are no other batcher instance present while GC cleanup.
784-
assertWithMessage("unexpected extra orphans").that(orphans).isEqualTo(0);
785-
Thread.sleep(DELAY_TIME * (1L << retry));
786-
}
787-
assertWithMessage("Batcher was not garbage collected").that(success).isTrue();
773+
await()
774+
.atMost(Duration.ofSeconds(5))
775+
.until(
776+
() -> {
777+
System.gc();
778+
System.runFinalization();
779+
int orphans = BatcherReference.cleanQueue();
780+
assertWithMessage("unexpected extra orphans").that(orphans).isAtMost(1);
781+
return orphans == 1;
782+
});
788783

789784
LogRecord lr;
790785
synchronized (records) {
@@ -809,15 +804,14 @@ void testClosedBatchersAreNotLogged() throws Exception {
809804
// Clean out the existing instances
810805
final long DELAY_TIME = 30L;
811806
int actualRemaining = 0;
812-
for (int retry = 0; retry < 3; retry++) {
813-
System.gc();
814-
System.runFinalization();
815-
actualRemaining = BatcherReference.cleanQueue();
816-
if (actualRemaining == 0) {
817-
break;
818-
}
819-
Thread.sleep(DELAY_TIME * (1L << retry));
820-
}
807+
await()
808+
.atMost(Duration.ofSeconds(5))
809+
.until(
810+
() -> {
811+
System.gc();
812+
System.runFinalization();
813+
return BatcherReference.cleanQueue() == 0;
814+
});
821815
assertThat(actualRemaining).isAtMost(0);
822816

823817
// Capture logs
@@ -849,12 +843,17 @@ public boolean isLoggable(LogRecord record) {
849843
}
850844
}
851845
// Run GC a few times to give the batchers a chance to be collected
852-
for (int retry = 0; retry < 100; retry++) {
853-
System.gc();
854-
System.runFinalization();
855-
BatcherReference.cleanQueue();
856-
Thread.sleep(10);
857-
}
846+
final AtomicInteger runs = new AtomicInteger(0);
847+
await()
848+
.pollInterval(Duration.ofMillis(10))
849+
.atMost(Duration.ofSeconds(2))
850+
.until(
851+
() -> {
852+
System.gc();
853+
System.runFinalization();
854+
BatcherReference.cleanQueue();
855+
return runs.incrementAndGet() >= 20;
856+
});
858857

859858
synchronized (records) {
860859
assertThat(records).isEmpty();
@@ -990,10 +989,12 @@ void testThrottlingBlocking() throws Exception {
990989
// resulting in a shorter total_throttled_time at the verification of throttledTime
991990
// at the end of the test.
992991
// https://github.com/googleapis/sdk-platform-java/issues/1193
993-
do {
994-
Thread.sleep(10);
995-
} while (batcherAddThreadHolder.isEmpty()
996-
|| batcherAddThreadHolder.get(0).getState() != Thread.State.WAITING);
992+
await()
993+
.atMost(Duration.ofSeconds(5))
994+
.until(
995+
() ->
996+
!batcherAddThreadHolder.isEmpty()
997+
&& batcherAddThreadHolder.get(0).getState() == Thread.State.WAITING);
997998

998999
long beforeGetCall = System.currentTimeMillis();
9991000
executor.submit(

‎sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/batching/Semaphore64Test.java‎

Lines changed: 9 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -29,12 +29,14 @@
2929
*/
3030
package com.google.api.gax.batching;
3131

32+
import static org.awaitility.Awaitility.await;
3233
import static org.junit.jupiter.api.Assertions.assertEquals;
3334
import static org.junit.jupiter.api.Assertions.assertFalse;
3435
import static org.junit.jupiter.api.Assertions.assertThrows;
3536
import static org.junit.jupiter.api.Assertions.assertTrue;
3637
import static org.junit.jupiter.api.Assertions.fail;
3738

39+
import java.time.Duration;
3840
import java.util.LinkedList;
3941
import java.util.List;
4042
import java.util.concurrent.TimeUnit;
@@ -68,8 +70,7 @@ void testBlocking() throws InterruptedException {
6870
Thread t = new Thread(() -> semaphore.acquire(1));
6971
t.start();
7072

71-
Thread.sleep(50);
72-
assertTrue(t.isAlive());
73+
await().atMost(Duration.ofSeconds(5)).until(() -> t.getState() == Thread.State.WAITING);
7374

7475
semaphore.release(1);
7576
t.join();
@@ -95,8 +96,7 @@ void testReducePermitLimitBlocking() throws InterruptedException {
9596
Thread t = new Thread(() -> semaphore.acquire(1));
9697
t.start();
9798

98-
Thread.sleep(50);
99-
assertTrue(t.isAlive());
99+
await().atMost(Duration.ofSeconds(5)).until(() -> t.getState() == Thread.State.WAITING);
100100

101101
semaphore.release(1);
102102
t.join();
@@ -124,17 +124,15 @@ void testAcquirePartialBlocking() throws Exception {
124124
Thread t1 = new Thread(() -> semaphore.acquire(1));
125125
t1.start();
126126
// wait for thread to start
127-
Thread.sleep(100);
128-
assertTrue(t1.isAlive());
127+
await().atMost(Duration.ofSeconds(5)).until(() -> t1.getState() == Thread.State.WAITING);
129128
semaphore.release(6);
130129
t1.join();
131130

132131
// now there should be 4 permits available, acquiring 6 should block
133132
Thread t2 = new Thread(() -> semaphore.acquirePartial(6));
134133
t2.start();
135134
// wait fo thread to start
136-
Thread.sleep(100);
137-
assertTrue(t2.isAlive());
135+
await().atMost(Duration.ofSeconds(5)).until(() -> t2.getState() == Thread.State.WAITING);
138136
// limit should still be 5 and get limit should not block
139137
assertEquals(5, semaphore.getPermitLimit());
140138
}
@@ -158,8 +156,7 @@ void testIncreasePermitLimitBlocking() throws Exception {
158156
Thread t = new Thread(() -> semaphore.acquire(1));
159157
t.start();
160158

161-
Thread.sleep(50);
162-
assertTrue(t.isAlive());
159+
await().atMost(Duration.ofSeconds(5)).until(() -> t.getState() == Thread.State.WAITING);
163160

164161
semaphore.increasePermitLimit(1);
165162
t.join();
@@ -208,8 +205,7 @@ void testReleaseWontOverflowBlocking() throws Exception {
208205
semaphore.release(10);
209206
Thread t = new Thread(() -> semaphore.acquire(11));
210207
t.start();
211-
Thread.sleep(100);
212-
assertTrue(t.isAlive());
208+
await().atMost(Duration.ofSeconds(5)).until(() -> t.getState() == Thread.State.WAITING);
213209
}
214210

215211
@Test
@@ -239,7 +235,6 @@ void testPermitLimitUnderflowBlocking() throws Exception {
239235
semaphore.release(10);
240236
Thread t = new Thread(() -> semaphore.acquire(11));
241237
t.start();
242-
Thread.sleep(100);
243-
assertTrue(t.isAlive());
238+
await().atMost(Duration.ofSeconds(5)).until(() -> t.getState() == Thread.State.WAITING);
244239
}
245240
}

‎sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/retrying/ScheduledRetryingExecutorTest.java‎

Lines changed: 32 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030
package com.google.api.gax.retrying;
3131

3232
import static com.google.api.gax.retrying.FailingCallable.FAST_RETRY_SETTINGS;
33+
import static org.awaitility.Awaitility.await;
3334
import static org.junit.jupiter.api.Assertions.assertEquals;
3435
import static org.junit.jupiter.api.Assertions.assertFalse;
3536
import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -41,6 +42,7 @@
4142
import com.google.api.core.NanoClock;
4243
import com.google.api.gax.retrying.FailingCallable.CustomException;
4344
import com.google.api.gax.rpc.testing.FakeCallContext;
45+
import java.time.Duration;
4446
import java.util.concurrent.CancellationException;
4547
import java.util.concurrent.ExecutionException;
4648
import java.util.concurrent.Executors;
@@ -99,26 +101,30 @@ void testSuccessWithFailuresPeekAttempt() throws Exception {
99101

100102
future.setAttemptFuture(executor.submit(future));
101103

102-
int failedAttempts = 0;
103-
while (!future.isDone()) {
104-
ApiFuture<String> attemptResult = future.peekAttemptResult();
105-
if (attemptResult != null) {
106-
assertTrue(attemptResult.isDone());
107-
assertFalse(attemptResult.isCancelled());
108-
try {
109-
attemptResult.get();
110-
} catch (ExecutionException e) {
111-
if (e.getCause() instanceof CustomException) {
112-
failedAttempts++;
113-
}
114-
}
115-
}
116-
Thread.sleep(0L, 100);
117-
}
104+
final int[] failedAttempts = {0};
105+
await()
106+
.pollInterval(Duration.ofMillis(2))
107+
.atMost(Duration.ofSeconds(5))
108+
.until(
109+
() -> {
110+
ApiFuture<String> attemptResult = future.peekAttemptResult();
111+
if (attemptResult != null) {
112+
assertTrue(attemptResult.isDone());
113+
assertFalse(attemptResult.isCancelled());
114+
try {
115+
attemptResult.get();
116+
} catch (ExecutionException e) {
117+
if (e.getCause() instanceof CustomException) {
118+
failedAttempts[0]++;
119+
}
120+
}
121+
}
122+
return future.isDone();
123+
});
118124

119125
assertFutureSuccess(future);
120126
assertEquals(15, future.getAttemptSettings().getAttemptCount());
121-
assertTrue(failedAttempts > 0);
127+
assertTrue(failedAttempts[0] > 0);
122128
}
123129
}
124130

@@ -260,9 +266,12 @@ void testCancelOuterFutureAfterStart() throws Exception {
260266
callable.setExternalFuture(future);
261267
future.setAttemptFuture(executor.submit(future));
262268

263-
// The test sleeps a duration long enough to ensure that the future has been submitted for
264-
// execution
265-
Thread.sleep(150L);
269+
await()
270+
.atMost(Duration.ofSeconds(5))
271+
.until(
272+
() ->
273+
future.getAttemptSettings() != null
274+
&& future.getAttemptSettings().getAttemptCount() > 0);
266275

267276
boolean res = future.cancel(false);
268277
assertTrue(res);
@@ -302,7 +311,9 @@ void testCancelProxiedFutureAfterStart() throws Exception {
302311
callable.setExternalFuture(future);
303312
future.setAttemptFuture(executor.submit(future));
304313

305-
Thread.sleep(50L);
314+
await()
315+
.atMost(Duration.ofSeconds(5))
316+
.until(() -> callable.getFirstAttemptFinishedLatch().getCount() == 0);
306317

307318
// Note that shutdownNow() will not cancel internal FutureTasks automatically, which
308319
// may potentially cause another thread handing on RetryingFuture#get() call forever.

0 commit comments

Comments
 (0)