Repeated messages of shared dispatcher #16802

    @Test(invocationCount = 10)
    public void test() throws Exception {
        @Cleanup final PulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl("http://127.0.0.1:8080").build();
        final String topicBase = "persistent://public/default/my-topic-" + System.currentTimeMillis();
        for (int i = 0; i < 3; i++) {
            admin.topics().createPartitionedTopic(topicBase + "-" + i, i + 2);
        }
        @Cleanup final PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://localhost:6650").build();
        final Consumer<byte[]> consumer = client.newConsumer()
                .topicsPattern(topicBase + ".*")
                .subscriptionName("sub")
                .receiverQueueSize(10)
                .subscriptionType(SubscriptionType.Shared)
                .subscribe();
        final int numMessages = 100;
        for (int i = 0; i < 3; i++) {
            final Producer<byte[]> producer = client.newProducer()
                    .topic(topicBase + "-" + i)
                    .messageRoutingMode(MessageRoutingMode.RoundRobinPartition)
                    .create();
            for (int j = 0; j < numMessages; j++) {
                producer.send(("msg-content-" + i + "-" + j).getBytes());
            }
        }
        int numReceived = 0;
        while (true) {
            final Message<byte[]> msg = consumer.receive(1, TimeUnit.SECONDS);
            if (msg == null) {
                break;
            }
            numReceived++;
        }
        Assert.assertEquals(numReceived, numMessages * 3);
    }