Skip to content

Commit 6c07bfa

Browse files
karthikkondapallykarthikkondapally
committed
moved semaphores to countdownlatch
1 parent 6519c77 commit 6c07bfa

3 files changed

Lines changed: 20 additions & 31 deletions

File tree

‎extended/src/test/java/io/kubernetes/client/extended/leaderelection/LeaderElectionTest.java‎

Lines changed: 11 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -13,8 +13,8 @@
1313
package io.kubernetes.client.extended.leaderelection;
1414

1515
import static java.util.concurrent.TimeUnit.SECONDS;
16-
import static org.junit.Assert.*;
17-
import static org.mockito.Mockito.*;
16+
import static org.junit.Assert.assertEquals;
17+
import static org.mockito.Mockito.when;
1818

1919
import io.kubernetes.client.openapi.ApiException;
2020
import java.net.HttpURLConnection;
@@ -25,7 +25,6 @@
2525
import java.util.concurrent.CountDownLatch;
2626
import java.util.concurrent.ExecutorService;
2727
import java.util.concurrent.Executors;
28-
import java.util.concurrent.Semaphore;
2928
import java.util.concurrent.atomic.AtomicReference;
3029
import java.util.concurrent.locks.ReentrantLock;
3130
import java.util.function.Consumer;
@@ -285,25 +284,21 @@ public void testLeaderElectionCaptureException() throws ApiException, Interrupte
285284
leaderElectionConfig.setLeaseDuration(Duration.ofMillis(1000));
286285
leaderElectionConfig.setRetryPeriod(Duration.ofMillis(200));
287286
leaderElectionConfig.setRenewDeadline(Duration.ofMillis(700));
288-
final Semaphore sem = new Semaphore(1);
287+
final CountDownLatch cLatch = new CountDownLatch(1);
289288
LeaderElector leaderElector =
290289
new LeaderElector(
291290
leaderElectionConfig,
292291
(t) -> {
293292
actualException.set(t);
294-
sem.release();
293+
cLatch.countDown();
295294
});
296295

297296
ExecutorService leaderElectionWorker = Executors.newFixedThreadPool(1);
298-
sem.acquire();
299297
leaderElectionWorker.submit(
300298
() -> {
301299
leaderElector.run(() -> {}, () -> {});
302300
});
303-
while (!sem.tryAcquire()) {
304-
System.out.println(
305-
"waiting for leaderElectionWorker to throw exception in LeaderElectionTest::testLeaderElectionCaptureException");
306-
}
301+
cLatch.await();
307302

308303
assertEquals(expectedException, actualException.get().getCause());
309304
}
@@ -341,21 +336,20 @@ public void testLeaderElectionReportLeaderOnStart() throws ApiException, Interru
341336
leaderElectionConfig.setRenewDeadline(Duration.ofMillis(700));
342337
LeaderElector leaderElector = new LeaderElector(leaderElectionConfig);
343338
ExecutorService leaderElectionWorker = Executors.newFixedThreadPool(1);
344-
final Semaphore s = new Semaphore(2);
345-
s.acquire(2);
339+
final CountDownLatch cLatch = new CountDownLatch(2);
346340
leaderElectionWorker.submit(
347341
() -> {
348342
leaderElector.run(
349343
() -> {},
350344
() -> {},
351345
(id) -> {
352346
notifications.add(id);
353-
s.release();
347+
cLatch.countDown();
354348
});
355349
});
356350

357351
// wait for two notifications to occur.
358-
s.acquire(2);
352+
cLatch.await();
359353

360354
assertEquals(2, notifications.size());
361355
assertEquals("foo2", notifications.get(0));
@@ -385,20 +379,19 @@ public void testLeaderElectionShouldReportLeaderItAcquiresOnStart()
385379
leaderElectionConfig.setRenewDeadline(Duration.ofMillis(700));
386380
LeaderElector leaderElector = new LeaderElector(leaderElectionConfig);
387381
ExecutorService leaderElectionWorker = Executors.newFixedThreadPool(1);
388-
Semaphore s = new Semaphore(1);
389-
s.acquire();
382+
final CountDownLatch cLatch = new CountDownLatch(1);
390383
leaderElectionWorker.submit(
391384
() -> {
392385
leaderElector.run(
393386
() -> {},
394387
() -> {},
395388
(id) -> {
396389
notifications.add(id);
397-
s.release();
390+
cLatch.countDown();
398391
});
399392
});
400393

401-
s.acquire();
394+
cLatch.await();
402395
assertEquals(1, notifications.size());
403396
assertEquals("foo1", notifications.get(0));
404397
}

‎extended/src/test/java/io/kubernetes/client/extended/workqueue/DefaultDelayingQueueTest.java‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -103,7 +103,6 @@ private boolean waitForAdded(DefaultDelayingQueue queue, int size) {
103103
}
104104

105105
private boolean waitForWaitingQueueToFill(DefaultDelayingQueue queue) {
106-
return Wait.poll(
107-
Duration.ofMillis(10), Duration.ofSeconds(10), () -> queue.waitingForAddQueue.size() == 0);
106+
return waitForAdded(queue, 0);
108107
}
109108
}

‎util/src/test/java/io/kubernetes/client/ExecTest.java‎

Lines changed: 8 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,7 @@
3434
import java.io.InputStream;
3535
import java.io.OutputStream;
3636
import java.nio.charset.StandardCharsets;
37-
import java.util.concurrent.Semaphore;
37+
import java.util.concurrent.CountDownLatch;
3838
import org.junit.Before;
3939
import org.junit.Rule;
4040
import org.junit.Test;
@@ -83,7 +83,8 @@ public static InputStream makeStream(byte[] prefix, byte[] data) {
8383
return new ByteArrayInputStream(out);
8484
}
8585

86-
public static Thread asyncCopy(final InputStream is, final OutputStream os, Semaphore sem) {
86+
public static Thread asyncCopy(
87+
final InputStream is, final OutputStream os, CountDownLatch cLatch) {
8788
Thread t =
8889
new Thread(
8990
new Runnable() {
@@ -93,7 +94,7 @@ public void run() {
9394
} catch (IOException ex) {
9495
ex.printStackTrace();
9596
} finally {
96-
sem.release();
97+
cLatch.countDown();
9798
}
9899
}
99100
});
@@ -113,17 +114,13 @@ public void testExecProcess() throws IOException, InterruptedException {
113114

114115
final ByteArrayOutputStream stdout = new ByteArrayOutputStream();
115116
final ByteArrayOutputStream stderr = new ByteArrayOutputStream();
116-
final Semaphore sem = new Semaphore(2);
117-
sem.acquire(2);
118-
Thread t1 = asyncCopy(process.getInputStream(), stdout, sem);
119-
Thread t2 = asyncCopy(process.getErrorStream(), stderr, sem);
117+
CountDownLatch cLatch = new CountDownLatch(2);
118+
Thread t1 = asyncCopy(process.getInputStream(), stdout, cLatch);
119+
Thread t2 = asyncCopy(process.getErrorStream(), stderr, cLatch);
120120

121121
process.getHandler().bytesMessage(makeStream(3, OUTPUT_EXIT0.getBytes(StandardCharsets.UTF_8)));
122122

123-
while (!sem.tryAcquire(2)) {
124-
System.out.println(
125-
"waiting for async Copy task to be completed in ExecTest::testExecProcess");
126-
}
123+
cLatch.await();
127124
process.destroy();
128125

129126
assertEquals(msgData, stdout.toString());

0 commit comments

Comments
 (0)