Apache
/
hadoop
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

HADOOP-17040. Fix intermittent failure of ITestBlockingThreadPoolExecutorService. (#2020) 

(cherry picked from commit 9685314633)
This commit is contained in:
Masatake Iwasaki 2020-05-22 18:50:19 +09:00
parent 6b54f259e7
commit 4d30c395f7
1 changed files with 24 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

37
hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestBlockingThreadPoolExecutorService.java
View File

@ -31,11 +31,11 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import java.util.concurrent.Callable; import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
/** /**
* Basic test for S3A's blocking executor service. * Basic test for S3A's blocking executor service.
@ -92,11 +92,12 @@ public class ITestBlockingThreadPoolExecutorService {
*/ */
protected void verifyQueueSize(ExecutorService executorService, protected void verifyQueueSize(ExecutorService executorService,
int expectedQueueSize) { int expectedQueueSize) {
StopWatch stopWatch = new StopWatch().start(); CountDownLatch latch = new CountDownLatch(1);
for (int i = 0; i < expectedQueueSize; i++) { for (int i = 0; i < expectedQueueSize; i++) {
executorService.submit(sleeper); executorService.submit(new LatchedSleeper(latch));
assertDidntBlock(stopWatch);
} }
StopWatch stopWatch = new StopWatch().start();
latch.countDown();
executorService.submit(sleeper); executorService.submit(sleeper);
assertDidBlock(stopWatch); assertDidBlock(stopWatch);
} }
@ -124,15 +125,6 @@ public class ITestBlockingThreadPoolExecutorService {
// Helper functions, etc. // Helper functions, etc.
private void assertDidntBlock(StopWatch sw) {
try {
assertFalse("Non-blocking call took too long.",
sw.now(TimeUnit.MILLISECONDS) > BLOCKING_THRESHOLD_MSEC);
} finally {
sw.reset().start();
}
}
private void assertDidBlock(StopWatch sw) { private void assertDidBlock(StopWatch sw) {
try { try {
if (sw.now(TimeUnit.MILLISECONDS) < BLOCKING_THRESHOLD_MSEC) { if (sw.now(TimeUnit.MILLISECONDS) < BLOCKING_THRESHOLD_MSEC) {
@ -164,6 +156,25 @@ public class ITestBlockingThreadPoolExecutorService {
} }
}; };
private class LatchedSleeper implements Runnable {
private final CountDownLatch latch;
LatchedSleeper(CountDownLatch latch) {
this.latch = latch;
}
@Override
public void run() {
try {
latch.await();
Thread.sleep(TASK_SLEEP_MSEC);
} catch (InterruptedException e) {
LOG.info("Thread {} interrupted.", Thread.currentThread().getName());
Thread.currentThread().interrupt();
}
}
}
/** /**
* Helper function to create thread pool under test. * Helper function to create thread pool under test.
*/ */