How to consume only the latest message published from the topic and ignore all previously published message in Pulsar

Lets say i have a publisher app and it runs and publishes 20 messages and then goes down

import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;

public class TestPub {
    public static void main(String[] args) throws PulsarClientException, InterruptedException {
        PulsarClient client = PulsarClient.builder()
                .serviceUrl("pulsar://172.20.170.147:6650")
                .build();
        Producer<byte[]> producer = client.newProducer()
                .topic("example-topic")
                .create();

        for (int i = 0; i < 20; i++)
         {
            String msg = ""+i;
            producer.newMessage()
                    .value(msg.getBytes())
                    .send();
            System.out.println("Sent:"+msg);
            Thread.sleep(1000);
        }
    }
}

and then i have a subscriber app which can use either a consumer or reader but it starts after the publisher is done, i need it to read only the latest message published by publisher before i thought this code should do it but somehow it is not working, any clues folks?

import org.apache.pulsar.client.api.*;

public class TestSub {
    public static void main(String[] args) throws PulsarClientException {
        PulsarClient client = PulsarClient.builder()
                .serviceUrl("pulsar://172.20.170.147:6650")
                .build();

        read(client);
        consume(client);
    }

    static void consume(PulsarClient client) throws PulsarClientException  {
        Consumer consumer = client.newConsumer()
                .topic("example-topic")
                .subscriptionName("my-subscription-consumer")
                .subscriptionType(SubscriptionType.Exclusive)
                .subscriptionInitialPosition(SubscriptionInitialPosition.Latest)
                .startMessageIdInclusive()
                .subscribe();
        //consumer.seek(MessageId.latest);
        while (true) {
        // Wait for a message
            Message msg = consumer.receive();
            try {
                System.out.println("Message consumed: " +
                        new String(msg.getData()));
                //consumer.acknowledge(msg);
            } catch (Exception e) {
                consumer.negativeAcknowledge(msg);
            }
        }
    }

    static void read(PulsarClient client) throws PulsarClientException  {
        Reader consumer = client.newReader()
                .topic("example-topic")
                .subscriptionName("my-subscription-reader")
                .startMessageIdInclusive()
                .startMessageId(MessageId.latest)
                .create()
                ;
        while (true) {
            // Wait for a message
            Message msg = consumer.readNext();
            try {
                System.out.println("Message read: " +
                        new String(msg.getData()));
                //consumer.acknowledge(msg);
            } catch (Exception e) {
                System.out.println(e);
            }
        }

    }

}