You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2022/04/10 07:09:52 UTC
[pulsar] branch master updated: [fix][test] Add receive timeout to avoid block thread. (#15088)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 960580764bb [fix][test] Add receive timeout to avoid block thread. (#15088)
960580764bb is described below
commit 960580764bbc55d91f06ba95ee4187128c6a7565
Author: Qiang Zhao <74...@users.noreply.github.com>
AuthorDate: Sun Apr 10 15:09:44 2022 +0800
[fix][test] Add receive timeout to avoid block thread. (#15088)
---
.../org/apache/pulsar/client/api/SimpleProducerConsumerTest.java | 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java
index f0e4cca9a25..879553ba6a1 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java
@@ -2436,7 +2436,7 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
Message<byte[]> msg;
List<Message<byte[]>> messages1 = Lists.newArrayList();
for (int i = 0; i < consumeMsgInParts; i++) {
- msg = consumer.receive();
+ msg = consumer.receive(RECEIVE_TIMEOUT_SECONDS, TimeUnit.SECONDS);
if (msg != null) {
messages1.add(msg);
consumer.acknowledge(msg);
@@ -2452,7 +2452,7 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
// (1.b) consume second consumeMsgInParts msgs and trigger redeliver
messages1.clear();
for (int i = 0; i < consumeMsgInParts; i++) {
- msg = consumer.receive();
+ msg = consumer.receive(RECEIVE_TIMEOUT_SECONDS, TimeUnit.SECONDS);
if (msg != null) {
messages1.add(msg);
consumer.acknowledge(msg);
@@ -2475,7 +2475,7 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
int remainingMsgs = (2 * receiverQueueSize) - (2 * consumeMsgInParts);
messages1.clear();
for (int i = 0; i < remainingMsgs; i++) {
- msg = consumer.receive();
+ msg = consumer.receive(RECEIVE_TIMEOUT_SECONDS, TimeUnit.SECONDS);
if (msg != null) {
messages1.add(msg);
consumer.acknowledge(msg);