@Test(dataProvider = "topicName")
    public void testSkipMessages(String topicName) throws Exception {
        final String subName = topicName;
        assertEquals(admin.topics().getList("prop-xyz/ns1"), new ArrayList<>());

        final String persistentTopicName = "persistent://prop-xyz/ns1/" + topicName;
        // Force to create a topic
        publishMessagesOnPersistentTopic("persistent://prop-xyz/ns1/" + topicName, 0);
        assertEquals(admin.topics().getList("prop-xyz/ns1"),
                List.of("persistent://prop-xyz/ns1/" + topicName));

        // create consumer and subscription
        @Cleanup
        PulsarClient client = PulsarClient.builder()
                .serviceUrl(pulsar.getWebServiceAddress())
                .statsInterval(0, TimeUnit.SECONDS)
                .build();
        AtomicInteger total = new AtomicInteger();
        Consumer<byte[]> consumer = client.newConsumer().topic(persistentTopicName)
                .messageListener(new MessageListener<byte[]>() {
                    @SneakyThrows
                    @Override
                    public void received(Consumer<byte[]> consumer, Message<byte[]> msg) {
                        if (total.get() %2 !=0){
                            consumer.acknowledge(msg);
                        }
                        total.incrementAndGet();
                    }
                })
                .subscriptionName(subName)
                .subscriptionType(SubscriptionType.Exclusive).subscribe();

        assertEquals(admin.topics().getSubscriptions(persistentTopicName), List.of(subName));

        publishMessagesOnPersistentTopic("persistent://prop-xyz/ns1/" + topicName, 100);
        TimeUnit.SECONDS.sleep(3);
        TopicStats topicStats = admin.topics().getStats(persistentTopicName);
        long msgBacklog = topicStats.getSubscriptions().get(subName).getMsgBacklog();
        log.info("back={}",msgBacklog);
        int skipNumber = 20;
        admin.topics().skipMessages(persistentTopicName, subName, skipNumber);
        topicStats = admin.topics().getStats(persistentTopicName);
        assertEquals(topicStats.getSubscriptions().get(subName).getMsgBacklog(), msgBacklog - skipNumber);
    }

        public void setStartPosition(PositionImpl startPosition) {
            PositionImpl cp = new PositionImpl(startPosition.ledgerId, startPosition.entryId);
            this.startPosition = cp;
        }

        public void setEndPosition(PositionImpl endPosition) {
            PositionImpl cp = new PositionImpl(endPosition.ledgerId, endPosition.entryId);
            this.endPosition = cp;
        }
	
	// state.endPosition = r.lowerEndpoint();
	state.setEndPosition(r.lowerEndpoint());

	// state.startPosition = r.upperEndpoint();
	state.setStartPosition(r.upperEndpoint());