peekMessages return Message not found

[hakangs]

Expected behavior

By using peekMessages, I want to retrieve last 10 messages of a topic. Here is the code snippet I am using:

private static void getLatestMessages() {
		try {
			PulsarAdmin admin = createAdmin();
			List<Message> messages = admin.persistentTopics().peekMessages(PERSISTENT_SAMPLE_STANDALONE_NS1_MY_TOPIC,
					"my-subscribtion-name", 10);
			for (Message msg : messages) {
				String currText = panel.getTextArea().getText();
				panel.getTextArea().setText(currText + "\n" + new String(msg.getData()));
			}
		} catch (MalformedURLException e1) {
			e1.printStackTrace();
		} catch (PulsarClientException e1) {
			e1.printStackTrace();
		} catch (PulsarAdminException e) {
			e.printStackTrace();
		}
	}

	public static PulsarAdmin createAdmin() throws MalformedURLException, PulsarClientException {
		URL url = new URL("http://localhost:8080");
		boolean useTls = false;
		boolean tlsAllowInsecureConnection = false;
		String tlsTrustCertsFilePath = null;
		ClientConfiguration config = new ClientConfiguration();
		config.setUseTls(useTls);
		config.setTlsAllowInsecureConnection(tlsAllowInsecureConnection);
		config.setTlsTrustCertsFilePath(tlsTrustCertsFilePath);
		return new PulsarAdmin(url, config);
	}

Actual behavior

com.yahoo.pulsar.client.admin.PulsarAdminException$NotFoundException: Message not found
	at com.yahoo.pulsar.client.admin.internal.BaseResource.getApiException(BaseResource.java:171)
	at com.yahoo.pulsar.client.admin.internal.PersistentTopicsImpl$8.completed(PersistentTopicsImpl.java:457)
	at com.yahoo.pulsar.client.admin.internal.PersistentTopicsImpl$8.completed(PersistentTopicsImpl.java:449)
	at org.glassfish.jersey.client.JerseyInvocation$7.completed(JerseyInvocation.java:946)
	at org.glassfish.jersey.client.ClientRuntime.processResponse(ClientRuntime.java:196)
	at org.glassfish.jersey.client.ClientRuntime.access$300(ClientRuntime.java:74)
	at org.glassfish.jersey.client.ClientRuntime$2$1$1.run(ClientRuntime.java:166)
	at org.glassfish.jersey.internal.Errors$1.call(Errors.java:271)
	at org.glassfish.jersey.internal.Errors$1.call(Errors.java:267)
	at org.glassfish.jersey.internal.Errors.process(Errors.java:315)
	at org.glassfish.jersey.internal.Errors.process(Errors.java:297)
	at org.glassfish.jersey.internal.Errors.process(Errors.java:267)
	at org.glassfish.jersey.process.internal.RequestScope.runInScope(RequestScope.java:340)
	at org.glassfish.jersey.client.ClientRuntime$2$1.response(ClientRuntime.java:164)
	at org.glassfish.jersey.client.internal.HttpUrlConnector$3.run(HttpUrlConnector.java:297)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	at jersey.repackaged.com.google.common.util.concurrent.MoreExecutors$DirectExecutorService.execute(MoreExecutors.java:299)
	at java.util.concurrent.AbstractExecutorService.submit(AbstractExecutorService.java:112)
	at jersey.repackaged.com.google.common.util.concurrent.AbstractListeningExecutorService.submit(AbstractListeningExecutorService.java:50)
	at jersey.repackaged.com.google.common.util.concurrent.AbstractListeningExecutorService.submit(AbstractListeningExecutorService.java:37)
	at org.glassfish.jersey.client.internal.HttpUrlConnector.apply(HttpUrlConnector.java:293)
	at org.glassfish.jersey.client.ClientRuntime$2.run(ClientRuntime.java:180)
	at org.glassfish.jersey.internal.Errors$1.call(Errors.java:271)
	at org.glassfish.jersey.internal.Errors$1.call(Errors.java:267)
	at org.glassfish.jersey.internal.Errors.process(Errors.java:315)
	at org.glassfish.jersey.internal.Errors.process(Errors.java:297)
	at org.glassfish.jersey.internal.Errors.process(Errors.java:267)
	at org.glassfish.jersey.process.internal.RequestScope.runInScope(RequestScope.java:340)
	at org.glassfish.jersey.client.ClientRuntime$3.run(ClientRuntime.java:208)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	at java.lang.Thread.run(Thread.java:745)
Caused by: javax.ws.rs.ClientErrorException: HTTP 404 Not Found
	at com.yahoo.pulsar.client.admin.internal.PersistentTopicsImpl.getMessageFromHttpResponse(PersistentTopicsImpl.java:558)
	at com.yahoo.pulsar.client.admin.internal.PersistentTopicsImpl.access$000(PersistentTopicsImpl.java:58)
	at com.yahoo.pulsar.client.admin.internal.PersistentTopicsImpl$8.completed(PersistentTopicsImpl.java:454)
	... 33 more

Steps to reproduce

I am starting Pulsar as standalone, creating one producer and one consumer. I believe all previous messages are being persisted with Bookkeeper. What can be the reason that I can not fetch messages of the same topic ?

System configuration

Pulsar version: 1.16

[merlimat]

The peek-messages command will show messages that are in "backlog", meaning they were published but no consumer has either received them or acknowledged their consumption.

If you have the test consumer running, it will quickly receive and ack all the messages. Try to stop it, or make it consume messages slowly (eg: bin/pulsar-perf consume --rate 10) so that the backlog start accumulating. Then you can peek at messages

[hakangs]

thank you very much, I was acknowledging the messages...