This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 24fd30e1ac6 Stabilize query execution concurrency unit tests (#19079)
24fd30e1ac6 is described below
commit 24fd30e1ac6b43a32aaeca2066034b1ae81fbc44
Author: Caideyipi <[email protected]>
AuthorDate: Mon Nov 2 18:44:12 2026 +0800
Stabilize query execution concurrency unit tests (#19079)
* Stabilize query execution concurrency unit tests
* Stabilize blocked priority queue unit tests
* Propagate worker failures and guarantee unit test cleanup
---
.../execution/exchange/SharedTsBlockQueueTest.java | 165 +++++++--------------
.../fragment/FragmentInstanceExecutionTest.java | 5 +-
.../source/SeriesScanUtilCancellationTest.java | 56 +++++--
.../schedule/queue/L1PriorityQueueTest.java | 45 +++---
.../schedule/queue/L2PriorityQueueTest.java | 44 +++---
.../queue/MultilevelPriorityQueueTest.java | 43 ++----
6 files changed, 152 insertions(+), 206 deletions(-)
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/exchange/SharedTsBlockQueueTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/exchange/SharedTsBlockQueueTest.java
index 391393cf5fa..18a3094507e 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/exchange/SharedTsBlockQueueTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/exchange/SharedTsBlockQueueTest.java
@@ -29,15 +29,17 @@ import org.apache.iotdb.mpp.rpc.thrift.TFragmentInstanceId;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
-import org.apache.tsfile.external.commons.lang3.Validate;
import org.apache.tsfile.read.common.block.TsBlock;
import org.junit.Assert;
import org.junit.Test;
import org.mockito.Mockito;
+import java.util.concurrent.CompletionService;
+import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
-import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
import static
com.google.common.util.concurrent.MoreExecutors.newDirectExecutorService;
@@ -119,8 +121,8 @@ public class SharedTsBlockQueueTest {
Assert.assertTrue(addFuture.isDone());
}
- @Test(timeout = 15000L)
- public void concurrencyTest() {
+ @Test
+ public void concurrencyTest() throws Exception {
final String queryId = "q0";
final long mockTsBlockSize = 1024L * 1024L;
@@ -140,120 +142,57 @@ public class SharedTsBlockQueueTest {
queue.getCanAddTsBlock().set(null);
queue.setMaxBytesCanReserve(Long.MAX_VALUE);
+ // Reuse an immutable block so mock generation is outside the concurrency
deadline.
+ TsBlock block = Utils.createMockTsBlock(mockTsBlockSize);
+ int blockCount = 1000;
ExecutorService executor = Executors.newFixedThreadPool(2);
- AtomicReference<Integer> numOfTimesSenderBlocked = new
AtomicReference<>(0);
- AtomicReference<Integer> numOfTimesReceiverBlocked = new
AtomicReference<>(0);
- AtomicReference<Integer> numOfTsBlocksToSend = new AtomicReference<>(1000);
- AtomicReference<Integer> numOfTsBlocksToReceive = new
AtomicReference<>(1000);
- executor.submit(
- new SendTask(
- queue, mockTsBlockSize, numOfTsBlocksToSend,
numOfTimesSenderBlocked, executor));
- executor.submit(
- new ReceiveTask(queue, numOfTsBlocksToReceive,
numOfTimesReceiverBlocked, executor));
-
- while (numOfTsBlocksToSend.get() != 0 && numOfTsBlocksToReceive.get() !=
0) {
- String message =
- String.format(
- "Sender %d: %d, Receiver %d: %d",
- numOfTimesSenderBlocked.get(),
- numOfTsBlocksToSend.get(),
- numOfTimesReceiverBlocked.get(),
- numOfTsBlocksToReceive.get());
- System.out.println(message);
- try {
- Thread.sleep(10L);
- } catch (InterruptedException e) {
- Assert.fail(e.getMessage());
- }
- }
- }
-
- private static class SendTask implements Runnable {
-
- private final SharedTsBlockQueue queue;
- private final long mockTsBlockSize;
- private final AtomicReference<Integer> numOfTsBlocksToSend;
- private final AtomicReference<Integer> numOfTimesBlocked;
- private final ExecutorService executor;
-
- public SendTask(
- SharedTsBlockQueue queue,
- long mockTsBlockSize,
- AtomicReference<Integer> numOfTsBlocksToSend,
- AtomicReference<Integer> numOfTimesBlocked,
- ExecutorService executor) {
- this.queue = Validate.notNull(queue);
- Validate.isTrue(mockTsBlockSize > 0L);
- this.mockTsBlockSize = mockTsBlockSize;
- this.numOfTsBlocksToSend = Validate.notNull(numOfTsBlocksToSend);
- this.numOfTimesBlocked = Validate.notNull(numOfTimesBlocked);
- this.executor = Validate.notNull(executor);
- }
-
- @Override
- public void run() {
- ListenableFuture<Void> blockedOnMemory = null;
- while (numOfTsBlocksToSend.get() > 0) {
- synchronized (queue) {
- blockedOnMemory =
queue.add(Utils.createMockTsBlock(mockTsBlockSize));
- }
- numOfTsBlocksToSend.updateAndGet(v -> v - 1);
- if (!blockedOnMemory.isDone()) {
- break;
- }
- }
-
- if (blockedOnMemory != null) {
- numOfTimesBlocked.updateAndGet(v -> v + 1);
- blockedOnMemory.addListener(
- new SendTask(queue, mockTsBlockSize, numOfTsBlocksToSend,
numOfTimesBlocked, executor),
- executor);
- } else {
- synchronized (queue) {
- queue.setNoMoreTsBlocks(true);
- }
+ CompletionService<Void> completedTasks = new
ExecutorCompletionService<>(executor);
+ try {
+ completedTasks.submit(
+ () -> {
+ for (int i = 0; i < blockCount; i++) {
+ ListenableFuture<Void> blocked;
+ synchronized (queue) {
+ blocked = queue.add(block);
+ }
+ blocked.get(10, TimeUnit.SECONDS);
+ }
+ synchronized (queue) {
+ queue.setNoMoreTsBlocks(true);
+ }
+ return null;
+ });
+ completedTasks.submit(
+ () -> {
+ for (int i = 0; i < blockCount; i++) {
+ ListenableFuture<Void> blocked;
+ synchronized (queue) {
+ blocked = queue.isBlocked();
+ }
+ blocked.get(10, TimeUnit.SECONDS);
+ synchronized (queue) {
+ Assert.assertSame(block, queue.remove());
+ }
+ }
+ return null;
+ });
+ for (int i = 0; i < 2; i++) {
+ Future<Void> completed = completedTasks.poll(10, TimeUnit.SECONDS);
+ Assert.assertNotNull(completed);
+ completed.get();
}
- }
- }
-
- private static class ReceiveTask implements Runnable {
-
- private final SharedTsBlockQueue queue;
- private final AtomicReference<Integer> numOfTsBlocksToReceive;
- private final AtomicReference<Integer> numOfTimesBlocked;
- private final ExecutorService executor;
-
- public ReceiveTask(
- SharedTsBlockQueue queue,
- AtomicReference<Integer> numOfTsBlocksToReceive,
- AtomicReference<Integer> numOfTimesBlocked,
- ExecutorService executor) {
- this.queue = Validate.notNull(queue);
- this.numOfTsBlocksToReceive = Validate.notNull(numOfTsBlocksToReceive);
- this.numOfTimesBlocked = Validate.notNull(numOfTimesBlocked);
- this.executor = Validate.notNull(executor);
- }
-
- @Override
- public void run() {
- ListenableFuture<Void> blocked = null;
- while (numOfTsBlocksToReceive.get() > 0) {
+ Assert.assertTrue(queue.hasNoMoreTsBlocks());
+ Assert.assertTrue(queue.isEmpty());
+ Assert.assertEquals(0, spyMemoryPool.getReservedBytes());
+ } finally {
+ executor.shutdownNow();
+ try {
+ Assert.assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS));
+ } finally {
synchronized (queue) {
- blocked = queue.isBlocked();
- if (blocked.isDone()) {
- queue.remove();
- numOfTsBlocksToReceive.updateAndGet(v -> v - 1);
- } else {
- break;
- }
+ queue.close();
}
}
-
- if (blocked != null) {
- numOfTimesBlocked.updateAndGet(v -> v + 1);
- blocked.addListener(
- new ReceiveTask(queue, numOfTsBlocksToReceive, numOfTimesBlocked,
executor), executor);
- }
}
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceExecutionTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceExecutionTest.java
index 0caf1692f63..237aacd72e4 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceExecutionTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceExecutionTest.java
@@ -134,7 +134,7 @@ public class FragmentInstanceExecutionTest {
}
@Test
- public void testTVListOwnerTransfer() throws InterruptedException {
+ public void testTVListOwnerTransfer() throws Exception {
ExecutorService instanceNotificationExecutor =
IoTDBThreadPoolFactory.newFixedThreadPool(1,
"test-instance-notification");
@@ -173,8 +173,9 @@ public class FragmentInstanceExecutionTest {
fragmentInstanceContext2.decrementNumOfUnClosedDriver();
fragmentInstanceContext1.getStateMachine().finished();
- Thread.sleep(100);
+ instanceNotificationExecutor.submit(() -> {}).get(10, TimeUnit.SECONDS);
fragmentInstanceContext2.getStateMachine().finished();
+ instanceNotificationExecutor.submit(() -> {}).get(10, TimeUnit.SECONDS);
assertTrue(execution1.getInstanceState().isDone());
assertTrue(execution2.getInstanceState().isDone());
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtilCancellationTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtilCancellationTest.java
index e347699cb6d..d0bda579aca 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtilCancellationTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtilCancellationTest.java
@@ -244,12 +244,12 @@ public class SeriesScanUtilCancellationTest {
assertEquals(2, loadedFiles);
}
- @Test(timeout = 15000)
+ @Test
public void testTimeoutUnblocksDriverResourceCleanup() throws Exception {
assertFailureUnblocksDriverResourceCleanup(new QueryTimeoutException());
}
- @Test(timeout = 15000)
+ @Test
public void testSemanticFailureUnblocksDriverResourceCleanup() throws
Exception {
assertFailureUnblocksDriverResourceCleanup(
new SemanticException("Scalar sub-query has returned multiple rows."));
@@ -260,6 +260,7 @@ public class SeriesScanUtilCancellationTest {
ExecutorService notifications = Executors.newSingleThreadExecutor();
FragmentInstanceContext context = newContext(notifications);
context.initializeNumOfDrivers(1);
+ ExecutorService worker = Executors.newSingleThreadExecutor();
try {
CountDownLatch closeRequested = new CountDownLatch(1);
SeriesScanUtil scanner =
@@ -314,10 +315,14 @@ public class SeriesScanUtilCancellationTest {
false,
exchangeManager);
- assertSame(
- failure,
- assertThrows(
- RuntimeException.class, () -> driver.processFor(new Duration(1,
TimeUnit.SECONDS))));
+ // Bound the driver operation rather than cold context and mock
initialization in a CI fork.
+ Future<RuntimeException> execution =
+ worker.submit(
+ () ->
+ assertThrows(
+ RuntimeException.class,
+ () -> driver.processFor(new Duration(1,
TimeUnit.SECONDS))));
+ assertSame(failure, execution.get(5, TimeUnit.SECONDS));
// This can complete only after the preceding cleanup callback gets past
allDriversClosed.
notifications.submit(() -> {}).get(5, TimeUnit.SECONDS);
assertEquals(2, loadedFiles);
@@ -329,12 +334,17 @@ public class SeriesScanUtilCancellationTest {
} finally {
// Ensure a failed assertion cannot leave the notification thread
waiting on the latch.
context.decrementNumOfUnClosedDriver();
- notifications.shutdownNow();
- assertTrue(notifications.awaitTermination(5, TimeUnit.SECONDS));
+ worker.shutdownNow();
+ notifications.shutdown();
+ try {
+ assertTrue(worker.awaitTermination(5, TimeUnit.SECONDS));
+ } finally {
+ assertTrue(notifications.awaitTermination(5, TimeUnit.SECONDS));
+ }
}
}
- @Test(timeout = 60000)
+ @Test
public void testFinishedScanUnblocksDriverResourceCleanup() throws Exception
{
for (boolean sequence : new boolean[] {true, false}) {
for (boolean waitForClose : new boolean[] {false, true}) {
@@ -352,7 +362,6 @@ public class SeriesScanUtilCancellationTest {
CountDownLatch resumeScan = new CountDownLatch(1);
CountDownLatch closeRequested = new CountDownLatch(1);
// Cover both orderings: the FI is FINISHED before driver.close(), and
close is already pending.
- notifications.submit(() -> await(allowNotifications));
FragmentInstanceContext context = newContext(notifications);
context.initializeNumOfDrivers(1);
Operator operator = mock(Operator.class);
@@ -401,6 +410,13 @@ public class SeriesScanUtilCancellationTest {
false,
exchangeManager);
try {
+ // Release this gate in finally so a timeout cannot change the ordering
under test.
+ Future<?> notificationsAllowed =
+ notifications.submit(
+ () -> {
+ allowNotifications.await();
+ return null;
+ });
Future<?> execution =
worker.submit(() -> driver.processFor(new Duration(1,
TimeUnit.SECONDS)));
await(scanEntered);
@@ -415,6 +431,7 @@ public class SeriesScanUtilCancellationTest {
assertEquals(FragmentInstanceState.FINISHED,
context.getStateMachine().getState());
assertTrue(context.getStateMachine().getFailureCauses().isEmpty());
allowNotifications.countDown();
+ notificationsAllowed.get(5, TimeUnit.SECONDS);
notifications.submit(() -> {}).get(5, TimeUnit.SECONDS);
verify(operator).close();
verify(context.getMemoryReservationContext()).releaseAllReservedMemory();
@@ -426,12 +443,19 @@ public class SeriesScanUtilCancellationTest {
resumeScan.countDown();
allowNotifications.countDown();
worker.shutdownNow();
- assertTrue(worker.awaitTermination(5, TimeUnit.SECONDS));
- driver.close();
- context.decrementNumOfUnClosedDriver();
- // Let FI cleanup finish after opening its latch, without interrupting
its driver-close wait.
- notifications.shutdown();
- assertTrue(notifications.awaitTermination(5, TimeUnit.SECONDS));
+ try {
+ assertTrue(worker.awaitTermination(5, TimeUnit.SECONDS));
+ } finally {
+ try {
+ driver.close();
+ } finally {
+ context.decrementNumOfUnClosedDriver();
+ // Let FI cleanup finish after opening its latch, without
interrupting its driver-close
+ // wait.
+ notifications.shutdown();
+ assertTrue(notifications.awaitTermination(5, TimeUnit.SECONDS));
+ }
+ }
}
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/schedule/queue/L1PriorityQueueTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/schedule/queue/L1PriorityQueueTest.java
index 4038bd51c28..cb5abc56abb 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/schedule/queue/L1PriorityQueueTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/schedule/queue/L1PriorityQueueTest.java
@@ -23,13 +23,15 @@ import
org.apache.iotdb.calc.execution.schedule.queue.IndexedBlockingQueue;
import org.junit.Assert;
import org.junit.Test;
-import java.util.ArrayList;
-import java.util.List;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.TimeUnit;
+
+import static org.awaitility.Awaitility.await;
public class L1PriorityQueueTest {
@Test
- public void testPollBlocked() throws InterruptedException {
+ public void testPollBlocked() throws Exception {
IndexedBlockingQueue<QueueElement> queue =
new L1PriorityQueue<>(
10,
@@ -40,27 +42,24 @@ public class L1PriorityQueueTest {
return Integer.compare(o1.getValue(), o2.getValue());
},
new QueueElement(new QueueElement.QueueElementID(0), 0));
- List<QueueElement> res = new ArrayList<>();
- Thread t1 =
- new Thread(
- () -> {
- try {
- QueueElement e = queue.poll();
- res.add(e);
- } catch (InterruptedException e) {
- e.printStackTrace();
- Assert.fail();
- }
- });
+ FutureTask<QueueElement> result = new FutureTask<>(queue::poll);
+ Thread t1 = new Thread(result);
t1.start();
- Thread.sleep(100);
- Assert.assertEquals(Thread.State.WAITING, t1.getState());
- QueueElement e2 = new QueueElement(new QueueElement.QueueElementID(1), 1);
- queue.push(e2);
- Thread.sleep(100);
- Assert.assertEquals(Thread.State.TERMINATED, t1.getState());
- Assert.assertEquals(1, res.size());
- Assert.assertEquals(e2.getDriverTaskId().toString(),
res.get(0).getDriverTaskId().toString());
+ try {
+ await()
+ .atMost(10, TimeUnit.SECONDS)
+ .until(() -> result.isDone() || t1.getState() ==
Thread.State.WAITING);
+ if (result.isDone()) {
+ Assert.fail("poll completed before an element was pushed: " +
result.get());
+ }
+ QueueElement e2 = new QueueElement(new QueueElement.QueueElementID(1),
1);
+ queue.push(e2);
+ Assert.assertSame(e2, result.get(10, TimeUnit.SECONDS));
+ } finally {
+ t1.interrupt();
+ t1.join(TimeUnit.SECONDS.toMillis(10));
+ Assert.assertFalse(t1.isAlive());
+ }
}
@Test
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/schedule/queue/L2PriorityQueueTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/schedule/queue/L2PriorityQueueTest.java
index 537358d9a49..3a9e85fcc2f 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/schedule/queue/L2PriorityQueueTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/schedule/queue/L2PriorityQueueTest.java
@@ -23,8 +23,7 @@ import
org.apache.iotdb.calc.execution.schedule.queue.IndexedBlockingQueue;
import org.junit.Assert;
import org.junit.Test;
-import java.util.ArrayList;
-import java.util.List;
+import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;
import static org.awaitility.Awaitility.await;
@@ -32,7 +31,7 @@ import static org.junit.Assert.fail;
public class L2PriorityQueueTest {
@Test
- public void testPollBlocked() throws InterruptedException {
+ public void testPollBlocked() throws Exception {
IndexedBlockingQueue<QueueElement> queue =
new L2PriorityQueue<>(
10,
@@ -43,29 +42,24 @@ public class L2PriorityQueueTest {
return Integer.compare(o1.getValue(), o2.getValue());
},
new QueueElement(new QueueElement.QueueElementID(0), 0));
- List<QueueElement> res = new ArrayList<>();
- Thread t1 =
- new Thread(
- () -> {
- try {
- QueueElement e = queue.poll();
- res.add(e);
- } catch (InterruptedException e) {
- e.printStackTrace();
- fail();
- }
- });
+ FutureTask<QueueElement> result = new FutureTask<>(queue::poll);
+ Thread t1 = new Thread(result);
t1.start();
- await()
- .atMost(1, TimeUnit.MINUTES)
- .untilAsserted(() -> Assert.assertEquals(Thread.State.WAITING,
t1.getState()));
- QueueElement e2 = new QueueElement(new QueueElement.QueueElementID(1), 1);
- queue.push(e2);
- await()
- .atMost(1, TimeUnit.MINUTES)
- .untilAsserted(() -> Assert.assertEquals(Thread.State.TERMINATED,
t1.getState()));
- Assert.assertEquals(1, res.size());
- Assert.assertEquals(e2.getDriverTaskId().toString(),
res.get(0).getDriverTaskId().toString());
+ try {
+ await()
+ .atMost(10, TimeUnit.SECONDS)
+ .until(() -> result.isDone() || t1.getState() ==
Thread.State.WAITING);
+ if (result.isDone()) {
+ Assert.fail("poll completed before an element was pushed: " +
result.get());
+ }
+ QueueElement e2 = new QueueElement(new QueueElement.QueueElementID(1),
1);
+ queue.push(e2);
+ Assert.assertSame(e2, result.get(10, TimeUnit.SECONDS));
+ } finally {
+ t1.interrupt();
+ t1.join(TimeUnit.SECONDS.toMillis(10));
+ Assert.assertFalse(t1.isAlive());
+ }
}
@Test
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/schedule/queue/MultilevelPriorityQueueTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/schedule/queue/MultilevelPriorityQueueTest.java
index 42bb75a2dc5..7d9de811af5 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/schedule/queue/MultilevelPriorityQueueTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/schedule/queue/MultilevelPriorityQueueTest.java
@@ -35,44 +35,33 @@ import org.junit.Assert;
import org.junit.Test;
import org.mockito.Mockito;
-import java.util.ArrayList;
-import java.util.List;
import java.util.OptionalInt;
+import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;
import static org.awaitility.Awaitility.await;
public class MultilevelPriorityQueueTest {
@Test
- public void testPollBlocked() {
+ public void testPollBlocked() throws Exception {
+ IndexedBlockingQueue<DriverTask> queue = new MultilevelPriorityQueue(2,
1000, new DriverTask());
+ FutureTask<DriverTask> result = new FutureTask<>(queue::poll);
+ Thread t1 = new Thread(result);
+ t1.start();
try {
- IndexedBlockingQueue<DriverTask> queue =
- new MultilevelPriorityQueue(2, 1000, new DriverTask());
- List<DriverTask> res = new ArrayList<>();
- Thread t1 =
- new Thread(
- () -> {
- try {
- DriverTask e = queue.poll();
- res.add(e);
- } catch (InterruptedException e) {
- e.printStackTrace();
- Assert.fail();
- }
- });
- t1.start();
await()
- .atMost(1, TimeUnit.MINUTES)
- .untilAsserted(() -> Assert.assertEquals(Thread.State.WAITING,
t1.getState()));
+ .atMost(10, TimeUnit.SECONDS)
+ .until(() -> result.isDone() || t1.getState() ==
Thread.State.WAITING);
+ if (result.isDone()) {
+ Assert.fail("poll completed before an element was pushed: " +
result.get());
+ }
DriverTask e2 = mockDriverTask(mockDriverTaskId(), false);
queue.push(e2);
- await()
- .atMost(1, TimeUnit.MINUTES)
- .untilAsserted(() -> Assert.assertEquals(Thread.State.TERMINATED,
t1.getState()));
- Assert.assertEquals(1, res.size());
- Assert.assertEquals(e2.getDriverTaskId().toString(),
res.get(0).getDriverTaskId().toString());
- } catch (Exception e) {
- Assert.fail();
+ Assert.assertSame(e2, result.get(10, TimeUnit.SECONDS));
+ } finally {
+ t1.interrupt();
+ t1.join(TimeUnit.SECONDS.toMillis(10));
+ Assert.assertFalse(t1.isAlive());
}
}
