Avoid repeated calls to skipBrokerEntryMetadataIfExist when parseMess…

…ageMetadata (#767) Related to apache/pulsar#10968
streamnative · Sep 30, 2021 · 6c3370b · 6c3370b
1 parent c3bd3dd
commit 6c3370b
Show file tree

Hide file tree

Showing 2 changed files with 0 additions and 2 deletions.
diff --git a/...a-impl/src/main/java/io/streamnative/pulsar/handlers/kop/format/PulsarEntryFormatter.java b/...a-impl/src/main/java/io/streamnative/pulsar/handlers/kop/format/PulsarEntryFormatter.java
@@ -111,7 +111,6 @@ public DecodeResult decode(final List<Entry> entries, final byte magic) {
                 // each entry is a batched message
                 ByteBuf metadataAndPayload = entry.getDataBuffer();
 
-                Commands.skipBrokerEntryMetadataIfExist(metadataAndPayload);
                 MessageMetadata msgMetadata = Commands.parseMessageMetadata(metadataAndPayload);
 
                 decodeResults.add(ByteBufUtils.decodePulsarEntryToKafkaRecords(

diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/MessageIdUtils.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/MessageIdUtils.java
@@ -48,7 +48,6 @@ public static long getLogEndOffset(ManagedLedger managedLedger) {
 
     public static long getPublishTime(final ByteBuf byteBuf) {
         final int readerIndex = byteBuf.readerIndex();
-        Commands.skipBrokerEntryMetadataIfExist(byteBuf);
         final MessageMetadata metadata = Commands.parseMessageMetadata(byteBuf);
         byteBuf.readerIndex(readerIndex);
         return metadata.getPublishTime();