@Test
    public void testReplicatedSubscriptionAcrossTwoRegionsGetLastMessage() throws Exception {
        String namespace = BrokerTestUtil.newUniqueName("pulsar/replicatedsubscriptionlastmessage");
        String topicName = "persistent://" + namespace + "/mytopic";
        String subscriptionName = "cluster-subscription";
        // this setting can be used to manually run the test with subscription replication disabled
        // it shows that subscription replication has no impact in behavior for this test case
        boolean replicateSubscriptionState = true;

        admin1.namespaces().createNamespace(namespace);
        admin1.namespaces().setNamespaceReplicationClusters(namespace, Sets.newHashSet("r1", "r2"));

        @Cleanup
        PulsarClient client1 = PulsarClient.builder().serviceUrl(url1.toString())
                .statsInterval(0, TimeUnit.SECONDS)
                .build();

        // create subscription in r1
        createReplicatedSubscription(client1, topicName, subscriptionName, replicateSubscriptionState);

        @Cleanup
        PulsarClient client2 = PulsarClient.builder().serviceUrl(url2.toString())
                .statsInterval(0, TimeUnit.SECONDS)
                .build();

        // create subscription in r2
        createReplicatedSubscription(client2, topicName, subscriptionName, replicateSubscriptionState);

        Set<String> sentMessages = new LinkedHashSet<>();

        // send messages in r1
        @Cleanup
        Producer<byte[]> producer = client1.newProducer().topic(topicName)
                .enableBatching(false)
                .messageRoutingMode(MessageRoutingMode.SinglePartition)
                .create();
        int numMessages = 6;
        for (int i = 0; i < numMessages; i++) {
            String body = "message" + i;
            producer.send(body.getBytes(StandardCharsets.UTF_8));
            sentMessages.add(body);
        }
        producer.close();


        // consume 3 messages in r1
        Set<String> receivedMessages = new LinkedHashSet<>();
        try (Consumer<byte[]> consumer1 = client1.newConsumer()
                .topic(topicName)
                .subscriptionName(subscriptionName)
                .replicateSubscriptionState(replicateSubscriptionState)
                .subscribe()) {
            readMessages(consumer1, receivedMessages, 3, false);
        }

        // wait for subscription to be replicated
        Thread.sleep(2 * config1.getReplicatedSubscriptionsSnapshotFrequencyMillis());

        // create a reader in r2
        Reader<byte[]> reader = client2.newReader().topic(topicName)
                .subscriptionName("new-sub")
                .startMessageId(MessageId.earliest)
                .create();
        int readNum = 0;
        while (reader.hasMessageAvailable()) {
            Message<byte[]> message = reader.readNext(10, TimeUnit.SECONDS);
            System.out.println("Receive message: " + new String(message.getValue()) + " msgId: " + message.getMessageId());
            assertNotNull(message);
            readNum++;
        }
        assertEquals(readNum, numMessages);
    }