Fix race condition in testRedeliveryOnBlockedDistpatcher
@@ -31,6 +31,7 @@ import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.Queue;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
@@ -40,23 +41,14 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Supplier;
import java.util.stream.Collectors;
-import com.google.common.collect.Sets;
import org.apache.pulsar.broker.namespace.NamespaceService;
import org.apache.pulsar.broker.service.BrokerService;
import org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
-import org.apache.pulsar.client.api.Consumer;
-import org.apache.pulsar.client.api.ConsumerConfiguration;
-import org.apache.pulsar.client.api.Message;
-import org.apache.pulsar.client.api.MessageId;
-import org.apache.pulsar.client.api.Producer;
-import org.apache.pulsar.client.api.ProducerConfiguration;
-import org.apache.pulsar.client.api.PulsarClientException;
-import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.client.impl.ConsumerImpl;
import org.apache.pulsar.client.impl.MessageIdImpl;
-import org.apache.pulsar.common.policies.data.SubscriptionStats;
import org.apache.pulsar.common.policies.data.PersistentTopicStats;
+import org.apache.pulsar.common.policies.data.SubscriptionStats;
import org.apache.pulsar.common.util.collections.ConcurrentOpenHashSet;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -69,6 +61,8 @@ import org.testng.collections.Lists;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.Maps;
import com.google.common.collect.Multimap;
+import com.google.common.collect.Queues;
+import com.google.common.collect.Sets;
public class DispatcherBlockConsumerTest extends ProducerConsumerBase {
private static final Logger log = LoggerFactory.getLogger(DispatcherBlockConsumerTest.class);
@@ -160,7 +154,7 @@ public class DispatcherBlockConsumerTest extends ProducerConsumerBase {
// try to consume remaining messages: broker may take time to deliver so, retry multiple time to consume
// all messages
- List<MessageId> result = Lists.newArrayList();
+ Queue<MessageId> result = Queues.newConcurrentLinkedQueue();
// expecting messages which are not received
int expectedRemainingMessages = totalProducedMsgs - messages.size();
CountDownLatch latch = new CountDownLatch(expectedRemainingMessages);
@@ -264,7 +258,7 @@ public class DispatcherBlockConsumerTest extends ProducerConsumerBase {
// (4) try to consume remaining messages: broker may take time to deliver so, retry multiple time to consume
// all messages
- List<MessageId> result = Lists.newArrayList();
+ Queue<MessageId> result = Queues.newConcurrentLinkedQueue();
CountDownLatch latch = new CountDownLatch(totalProducedMsgs);
for (int i = 0; i < consumers.length; i++) {
final int consumerCount = i;
@@ -495,7 +489,7 @@ public class DispatcherBlockConsumerTest extends ProducerConsumerBase {
// try to consume remaining messages: broker may take time to deliver so, retry multiple time to consume
// all messages
CountDownLatch latch = new CountDownLatch(remainingMessages);
- List<MessageId> consumedMessages = Lists.newArrayList();
+ Queue<MessageId> consumedMessages = Queues.newConcurrentLinkedQueue();
for (int i = 0; i < consumers.length; i++) {
final int counsumerIndex = i;
for (int j = 0; j < remainingMessages; j++) {
