You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ch...@apache.org on 2024/04/16 09:00:31 UTC
(kafka) branch trunk updated: MINOR: Replaced the while loop with TestUtils.waitForCondition (#15678)
This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new f7eb96230d5 MINOR: Replaced the while loop with TestUtils.waitForCondition (#15678)
f7eb96230d5 is described below
commit f7eb96230d5d66fbd6e6d9480359ab82f925c230
Author: John Yu <yu...@gmail.com>
AuthorDate: Tue Apr 16 17:00:25 2024 +0800
MINOR: Replaced the while loop with TestUtils.waitForCondition (#15678)
Reviewers: Luke Chen <sh...@gmail.com>, Chia-Ping Tsai <ch...@gmail.com>
---
.../kafka/server/AssignmentsManagerTest.java | 28 +++++++++++-----------
1 file changed, 14 insertions(+), 14 deletions(-)
diff --git a/server/src/test/java/org/apache/kafka/server/AssignmentsManagerTest.java b/server/src/test/java/org/apache/kafka/server/AssignmentsManagerTest.java
index 93f7a39cab4..a385e2d8c9c 100644
--- a/server/src/test/java/org/apache/kafka/server/AssignmentsManagerTest.java
+++ b/server/src/test/java/org/apache/kafka/server/AssignmentsManagerTest.java
@@ -169,9 +169,7 @@ public class AssignmentsManagerTest {
public void testAssignmentAggregation() throws InterruptedException {
CountDownLatch readyToAssert = new CountDownLatch(1);
doAnswer(invocation -> {
- if (readyToAssert.getCount() > 0) {
- readyToAssert.countDown();
- }
+ readyToAssert.countDown();
return null;
}).when(channelManager).sendRequest(any(AssignReplicasToDirsRequest.Builder.class),
any(ControllerRequestCompletionHandler.class));
@@ -181,10 +179,11 @@ public class AssignmentsManagerTest {
manager.onAssignment(new TopicIdPartition(TOPIC_1, 3), DIR_3, "testAssignmentAggregation", () -> { });
manager.onAssignment(new TopicIdPartition(TOPIC_1, 4), DIR_1, "testAssignmentAggregation", () -> { });
manager.onAssignment(new TopicIdPartition(TOPIC_2, 5), DIR_2, "testAssignmentAggregation", () -> { });
- while (!readyToAssert.await(1, TimeUnit.MILLISECONDS)) {
+ TestUtils.waitForCondition(() -> {
time.sleep(100);
manager.wakeup();
- }
+ return readyToAssert.await(1, TimeUnit.MILLISECONDS);
+ }, "Timed out waiting for AssignReplicasToDirsRequest to be sent.");
ArgumentCaptor<AssignReplicasToDirsRequest.Builder> captor =
ArgumentCaptor.forClass(AssignReplicasToDirsRequest.Builder.class);
@@ -210,9 +209,7 @@ public class AssignmentsManagerTest {
void testRequeuesFailedAssignmentPropagations() throws InterruptedException {
CountDownLatch readyToAssert = new CountDownLatch(5);
doAnswer(invocation -> {
- if (readyToAssert.getCount() > 0) {
- readyToAssert.countDown();
- }
+ readyToAssert.countDown();
if (readyToAssert.getCount() == 4) {
invocation.getArgument(1, ControllerRequestCompletionHandler.class).onTimeout();
manager.onAssignment(new TopicIdPartition(TOPIC_1, 2), DIR_3, "testRequeuesFailedAssignmentPropagations", () -> { });
@@ -253,10 +250,11 @@ public class AssignmentsManagerTest {
any(ControllerRequestCompletionHandler.class));
manager.onAssignment(new TopicIdPartition(TOPIC_1, 1), DIR_1, "testRequeuesFailedAssignmentPropagations", () -> { });
- while (!readyToAssert.await(1, TimeUnit.MILLISECONDS)) {
+ TestUtils.waitForCondition(() -> {
time.sleep(TimeUnit.SECONDS.toMillis(1));
manager.wakeup();
- }
+ return readyToAssert.await(1, TimeUnit.MILLISECONDS);
+ }, "Timed out waiting for AssignReplicasToDirsRequest to be sent.");
ArgumentCaptor<AssignReplicasToDirsRequest.Builder> captor =
ArgumentCaptor.forClass(AssignReplicasToDirsRequest.Builder.class);
@@ -304,10 +302,11 @@ public class AssignmentsManagerTest {
manager.onAssignment(new TopicIdPartition(TOPIC_1, i % 5), DIR_1, "testOnCompletion", readyToAssert::countDown);
}
- while (!readyToAssert.await(1, TimeUnit.MILLISECONDS)) {
+ TestUtils.waitForCondition(() -> {
time.sleep(TimeUnit.SECONDS.toMillis(1));
manager.wakeup();
- }
+ return readyToAssert.await(1, TimeUnit.MILLISECONDS);
+ }, "Timed out waiting for AssignReplicasToDirsRequest to be sent.");
}
private static ClientResponse buildSuccessfulResponse(AssignReplicasToDirsRequestData request) {
@@ -393,9 +392,10 @@ public class AssignmentsManagerTest {
for (int i = 0; i < 4; i++) {
manager.onAssignment(new TopicIdPartition(TOPIC_1, i), DIR_1, "testQueuedReplicaToDirAssignmentsMetric", () -> { });
}
- while (!readyToAssert.await(1, TimeUnit.MILLISECONDS)) {
+ TestUtils.waitForCondition(() -> {
time.sleep(100);
- }
+ return readyToAssert.await(1, TimeUnit.MILLISECONDS);
+ }, "Timed out waiting for AssignReplicasToDirsRequest to be sent.");
assertEquals(4, queuedReplicaToDirAssignments.value());
for (int i = 4; i < 8; i++) {