diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java index 240bef95b0727..8bf5a7c5b7b0a 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java @@ -2450,6 +2450,7 @@ private Response generateResponseWithEntry(Entry entry) throws IOException { ByteBuf metadataAndPayload = entry.getDataBuffer(); // moves the readerIndex to the payload + Commands.skipBrokerEntryMetadataIfExist(metadataAndPayload); MessageMetadata metadata = Commands.parseMessageMetadata(metadataAndPayload); ResponseBuilder responseBuilder = Response.ok(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerEntryMetadataE2ETest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerEntryMetadataE2ETest.java index 1954271376a86..5cbaf3de9ae86 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerEntryMetadataE2ETest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerEntryMetadataE2ETest.java @@ -18,9 +18,11 @@ */ package org.apache.pulsar.broker.service; +import java.util.List; import lombok.Cleanup; import org.apache.pulsar.client.api.Consumer; import org.apache.pulsar.client.api.Message; +import org.apache.pulsar.client.api.MessageId; import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.SubscriptionType; import org.assertj.core.util.Sets; @@ -90,4 +92,21 @@ public void testProduceAndConsume(SubscriptionType subType) throws Exception { Assert.assertEquals(messages, receives); } + + @Test(timeOut = 20000) + public void testPeekMessage() throws Exception { + final String topic = newTopicName(); + final String subscription = "my-sub"; + + @Cleanup + Producer producer = pulsarClient.newProducer() + .topic(topic) + .create(); + producer.newMessage().value("hello".getBytes()).send(); + + admin.topics().createSubscription(topic, subscription, MessageId.earliest); + final List> messages = admin.topics().peekMessages(topic, subscription, 1); + Assert.assertEquals(messages.size(), 1); + Assert.assertEquals(messages.get(0).getData(), "hello".getBytes()); + } }