Skip to content

fix: remove Thread.sleep() from DefaultWorkQueueTest #4910

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

Open
sarveshkaushal wants to merge 1 commit into
base: master
Choose a base branch
from
+71 −67
Open

fix: remove Thread.sleep() from DefaultWorkQueueTest #4910

Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
138 changes: 71 additions & 67 deletions extended/src/test/java/io/kubernetes/client/extended/workqueue/DefaultWorkQueueTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,22 +14,24 @@

import static org.assertj.core.api.Assertions.assertThat;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import org.junit.jupiter.api.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

class DefaultWorkQueueTest {

private static final Logger LOGGER = LoggerFactory.getLogger(DefaultWorkQueueTest.class);
private static final long TIMEOUT_SECONDS = 10;

@Test
void multiProducerAndConsumers() throws Exception {
DefaultWorkQueue<String> queue = new DefaultWorkQueue<>();
final int producerCount = 10;
final int consumerCount = 5;
Semaphore start = new Semaphore(0);
ConcurrentLinkedQueue<Throwable> failures = new ConcurrentLinkedQueue<>();

// Start producers
CountDownLatch producerLatch = new CountDownLatch(producerCount);
Expand All @@ -39,12 +41,13 @@ void multiProducerAndConsumers() throws Exception {
new Thread(
() -> {
try {
start.acquire();
for (int j = 0; j < 50; j++) {
queue.add(String.valueOf(num));
Thread.sleep(10);
}
} catch (Exception e) {
// empty body
} catch (InterruptedException e) {
failures.add(e);
Thread.currentThread().interrupt();
} finally {
producerLatch.countDown();
}
Expand All @@ -55,93 +58,94 @@ void multiProducerAndConsumers() throws Exception {
// Start consumers
CountDownLatch consumerLatch = new CountDownLatch(consumerCount);
for (int i = 0; i < consumerCount; i++) {
final int num = i;
Thread t =
new Thread(
() -> {
try {
start.acquire();
for (; ; ) {
String item = queue.get();
assertThat(item)
.withFailMessage("Got an item added after shutdown")
.isNotEqualTo("added after shutdown!");
if (item == null) {
return;
}

LOGGER.info("Worker {}: begin processing {}", num, item);
Thread.sleep(50);
LOGGER.info("Worker {}: done processing {}", num, item);
if ("added after shutdown!".equals(item)) {
failures.add(new AssertionError("Got an item added after shutdown"));
}
queue.done(item);
}
} catch (Exception e) {
// empty body
} catch (InterruptedException e) {
failures.add(e);
Thread.currentThread().interrupt();
} finally {
consumerLatch.countDown();
}
});
t.start();
}

producerLatch.await();
queue.shutDown();
start.release(producerCount + consumerCount);

boolean producersFinished;
try {
producersFinished = producerLatch.await(TIMEOUT_SECONDS, TimeUnit.SECONDS);
} finally {
queue.shutDown();
}
queue.add("added after shutdown!");
consumerLatch.await();

assertThat(producersFinished).as("producers finished").isTrue();
assertThat(consumerLatch.await(TIMEOUT_SECONDS, TimeUnit.SECONDS))
.as("consumers finished")
.isTrue();
assertThat(failures).isEmpty();
assertThat(queue.length()).isZero();
}

@Test
void addWhileProcessing() throws Exception {
DefaultWorkQueue<String> queue = new DefaultWorkQueue<>();
final int producerCount = 10;
final int consumerCount = 5;
Semaphore itemDequeued = new Semaphore(0);
Semaphore allowDone = new Semaphore(0);
CountDownLatch consumerFinished = new CountDownLatch(1);
AtomicReference<Throwable> failure = new AtomicReference<>();

// Start producers
CountDownLatch producerLatch = new CountDownLatch(producerCount);
for (int i = 0; i < producerCount; i++) {
final int num = i;
Thread t =
new Thread(
() -> {
queue.add(String.valueOf(num));
producerLatch.countDown();
});
t.start();
}
queue.add("foo");

// Start consumers
CountDownLatch consumerLatch = new CountDownLatch(consumerCount);
for (int i = 0; i < consumerCount; i++) {
Thread t =
new Thread(
() -> {
// Every worker will re-add every item up to two times.
// This tests the dirty-while-processing case.
Map<String, Integer> counters = new HashMap<>();
try {
for (; ; ) {
String item = queue.get();
if (item == null) {
return;
}
counters.putIfAbsent(item, 1);
counters.computeIfPresent(item, (s, integer) -> counters.get(s) + 1);
if (counters.get(item) < 2) {
queue.add(item);
}
queue.done(item);
}
} catch (Exception e) {
// empty body
} finally {
consumerLatch.countDown();
}
});
t.start();
Thread consumer =
new Thread(
() -> {
try {
String item = queue.get();
itemDequeued.release();
allowDone.acquire();
queue.done(item);
} catch (InterruptedException e) {
failure.set(e);
Thread.currentThread().interrupt();
} finally {
consumerFinished.countDown();
}
});
consumer.start();

boolean dequeued = itemDequeued.tryAcquire(TIMEOUT_SECONDS, TimeUnit.SECONDS);
try {
assertThat(dequeued).as("consumer dequeued the item").isTrue();
queue.add("foo");
} finally {
allowDone.release();
}

producerLatch.await();
queue.shutDown();
consumerLatch.await();
assertThat(consumerFinished.await(TIMEOUT_SECONDS, TimeUnit.SECONDS))
.as("consumer finished")
.isTrue();
assertThat(failure.get()).isNull();
assertThat(queue.length()).isEqualTo(1);

String item = queue.get();
assertThat(item).isEqualTo("foo");
queue.done(item);
assertThat(queue.length()).isZero();
}

@Test
Expand Down