You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by li...@apache.org on 2022/10/27 08:13:36 UTC
[rocketmq] branch develop updated: [ISSUE #5331]Fix ConsumeQueueTest stable sleep cause NPE. (#5397)
This is an automated email from the ASF dual-hosted git repository.
lizhanhui pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
The following commit(s) were added to refs/heads/develop by this push:
new c10812407 [ISSUE #5331]Fix ConsumeQueueTest stable sleep cause NPE. (#5397)
c10812407 is described below
commit c1081240787b58bca6a0d88f164727fb6c5fbc03
Author: echooymxq <ec...@gmail.com>
AuthorDate: Thu Oct 27 16:13:28 2022 +0800
[ISSUE #5331]Fix ConsumeQueueTest stable sleep cause NPE. (#5397)
* [ISSUE #5331]Fix ConsumeQueueTest stable sleep cause NPE.
* Use awaitability to achieve granular condition check
* Add a timeout duration for condition check
* Add a timeout duration for condition check
Co-authored-by: Li Zhanhui <li...@gmail.com>
---
store/BUILD.bazel | 1 +
.../apache/rocketmq/store/ConsumeQueueTest.java | 25 +++++++++++++++++++---
2 files changed, 23 insertions(+), 3 deletions(-)
diff --git a/store/BUILD.bazel b/store/BUILD.bazel
index b839e72de..0419984eb 100644
--- a/store/BUILD.bazel
+++ b/store/BUILD.bazel
@@ -48,6 +48,7 @@ java_library(
"@maven//:com_conversantmedia_disruptor",
"@maven//:io_openmessaging_storage_dledger",
"@maven//:org_apache_commons_commons_lang3",
+ "@maven//:com_google_guava_guava",
],
)
diff --git a/store/src/test/java/org/apache/rocketmq/store/ConsumeQueueTest.java b/store/src/test/java/org/apache/rocketmq/store/ConsumeQueueTest.java
index c805f078b..625956ad4 100644
--- a/store/src/test/java/org/apache/rocketmq/store/ConsumeQueueTest.java
+++ b/store/src/test/java/org/apache/rocketmq/store/ConsumeQueueTest.java
@@ -25,6 +25,8 @@ import java.net.SocketAddress;
import java.net.UnknownHostException;
import java.util.Map;
import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.TimeUnit;
+
import org.apache.rocketmq.common.BrokerConfig;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.message.MessageConst;
@@ -35,6 +37,7 @@ import org.apache.rocketmq.store.queue.ConsumeQueueInterface;
import org.apache.rocketmq.store.queue.CqUnit;
import org.apache.rocketmq.store.queue.ReferredIterator;
import org.apache.rocketmq.store.stats.BrokerStatsManager;
+import org.awaitility.Awaitility;
import org.junit.Assert;
import org.junit.Test;
import org.mockito.Mockito;
@@ -250,7 +253,13 @@ public class ConsumeQueueTest {
for (int i = 0; i < totalMessages; i++) {
putMsg(messageStore);
}
- Thread.sleep(5);
+
+
+ // Wait consume queue build finish.
+ final MessageStore store = messageStore;
+ Awaitility.with().pollInterval(100, TimeUnit.MILLISECONDS).await().timeout(1, TimeUnit.MINUTES).until(() -> {
+ return store.dispatchBehindBytes() == 0;
+ });
ConsumeQueueInterface cq = messageStore.getConsumeQueueTable().get(TOPIC).get(QUEUE_ID);
Method method = cq.getClass().getDeclaredMethod("putMessagePositionInfo", long.class, int.class, long.class, long.class);
@@ -292,7 +301,12 @@ public class ConsumeQueueTest {
for (int i = 0; i < totalMessages; i++) {
putMsgMultiQueue(messageStore);
}
- Thread.sleep(5);
+
+ // Wait consume queue build finish.
+ final MessageStore store = messageStore;
+ Awaitility.with().pollInterval(100, TimeUnit.MILLISECONDS).await().timeout(1, TimeUnit.MINUTES).until(() -> {
+ return store.dispatchBehindBytes() == 0;
+ });
ConsumeQueueInterface cq = messageStore.getConsumeQueueTable().get(TOPIC).get(QUEUE_ID);
Method method = ((ConsumeQueue) cq).getClass().getDeclaredMethod("putMessagePositionInfoWrapper", DispatchRequest.class);
@@ -340,7 +354,12 @@ public class ConsumeQueueTest {
for (int i = 0; i < totalMessages; i++) {
putMsgMultiQueue(messageStore);
}
- Thread.sleep(5);
+
+ // Wait consume queue build finish.
+ final MessageStore store = messageStore;
+ Awaitility.with().pollInterval(100, TimeUnit.MILLISECONDS).await().timeout(1, TimeUnit.MINUTES).until(() -> {
+ return store.dispatchBehindBytes() == 0;
+ });
ConsumeQueueInterface cq = messageStore.getConsumeQueueTable().get(TOPIC).get(QUEUE_ID);