[fix][broker]fix Repeated messages of shared dispatcher

Author: poorbarcode

Diff for: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageRedeliveryController.java

@@ -51,6 +51,10 @@ public void add(long ledgerId, long entryId, long stickyKeyHash) {
         messagesToRedeliver.add(ledgerId, entryId);
     }
 
+    public boolean contains(long ledgerId, long entryId) {
+        return messagesToRedeliver.contains(ledgerId, entryId);
+    }
+
     public void remove(long ledgerId, long entryId) {
         if (hashesToBeBlocked != null) {
             hashesToBeBlocked.remove(ledgerId, entryId);

Diff for: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java

@@ -583,6 +583,9 @@ protected synchronized boolean trySendMessagesToConsumers(ReadType readType, Lis
         if (needTrimAckedMessages()) {
             cursor.trimDeletedEntries(entries);
         }
+        if (readType == ReadType.Replay) {
+            entries.removeIf(entry -> !redeliveryMessages.contains(entry.getLedgerId(), entry.getEntryId()));
+        }
 
         int entriesToDispatch = entries.size();
         // Trigger read more messages

Diff for: pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java

@@ -2383,6 +2383,99 @@ public void testPriorityConsumer() throws Exception {
         log.info("-- Exiting {} test --", methodName);
     }
 
+    @Test(timeOut = 30000)
+    public void concurrentlyReceiveDeliveriedMessages() throws Exception {
+        String topic = "persistent://my-property/my-ns/tp-" + UUID.randomUUID().toString();
+        String subName = "sub";
+        final int totalPublishMessages = 50;
+        // Make many messages in delivery queue.
+        Producer<byte[]> producer = pulsarClient.newProducer().topic(topic)
+                .enableBatching(false)
+                .messageRoutingMode(MessageRoutingMode.SinglePartition)
+                .create();
+        PulsarClient newPulsarClient = newPulsarClient(lookupUrl.toString(), 0);
+        Consumer<byte[]> consumer = newPulsarClient.newConsumer()
+                .topic(topic).subscriptionName(subName)
+                .subscriptionType(SubscriptionType.Shared).receiverQueueSize(10)
+                .acknowledgmentGroupTime(0, TimeUnit.SECONDS).subscribe();
+        for (int i = 0; i < totalPublishMessages; i++) {
+            final String message = "my-message-" + i;
+            producer.send(message.getBytes());
+        }
+        consumer.receive();
+        consumer.redeliverUnacknowledgedMessages();
+        consumer.close();
+
+        // create consumer tasks.
+        List<ConsumerTask> taskList = new ArrayList<>(10);
+        List<Thread> threadList = new ArrayList<>(10);
+        for (int i = 0; i < 10; i++){
+            PulsarClient pulsarClient = newPulsarClient(lookupUrl.toString(), 0);
+            ConsumerTask consumerTask = new ConsumerTask(pulsarClient, topic, subName);
+            taskList.add(consumerTask);
+            threadList.add(new Thread(consumerTask, consumerTask.consumerName));
+        }
+        // wait tasks finished.
+        threadList.forEach(thread -> thread.start());
+        for (Thread t : threadList){
+            t.join();
+        }
+        // Verify.
+        for (ConsumerTask task : taskList){
+            assertFalse(task.fail);
+        }
+        Assert.assertEquals(taskList.stream().map(t -> t.receivedEntries.size())
+                .reduce((a, b) -> a + b).get().intValue(), totalPublishMessages);
+        // cleanup.
+        producer.close();
+        admin.topics().delete(topic, false);
+    }
+
+    private static class ConsumerTask implements Runnable {
+
+        private static final AtomicInteger consumerNamSequence = new AtomicInteger();
+
+        private final List<Long> receivedEntries = new ArrayList<>();
+        private final PulsarClient pulsarClient;
+        private final String consumerName;
+        private final String topicName;
+        private final String subName;
+        private volatile boolean fail = false;
+
+        public ConsumerTask(PulsarClient newPulsarClient, String topicName, String subName){
+            this.pulsarClient = newPulsarClient;
+            this.topicName = topicName;
+            this.subName = subName;
+            this.consumerName = subName + "-" + consumerNamSequence.incrementAndGet();
+        }
+
+        @Override
+        public void run() {
+            try {
+                Consumer<byte[]> consumer = pulsarClient.newConsumer()
+                        .topic(topicName).subscriptionName(subName)
+                        .subscriptionType(SubscriptionType.Shared).receiverQueueSize(10)
+                        .consumerName(consumerName)
+                        .acknowledgmentGroupTime(0, TimeUnit.SECONDS).subscribe();
+                while(true){
+                    Message message = consumer.receive(2, TimeUnit.SECONDS);
+                    if (message == null){
+                        break;
+                    }
+                    MessageIdImpl messageId = (MessageIdImpl)message.getMessageId();
+                    consumer.acknowledge(messageId);
+                    System.out.println("===>12 GET " + messageId.getLedgerId() + ":" + messageId.getEntryId() + ", Consumer: " + consumerName);
+                    receivedEntries.add(messageId.getEntryId());
+                }
+                consumer.close();
+                pulsarClient.close();
+            } catch (Exception e) {
+                fail = true;
+                throw new RuntimeException(e);
+            }
+        }
+    }
+
     /**
      * <pre>
      * Verifies Dispatcher dispatches messages properly with shared-subscription consumers with combination of blocked