simplify peekOffsetFromEntry (apache#328)

simplify `MessageIdUtils.peekOffsetFromEntry` by use `Commands.peekBrokerEntryMetadataIfExist` directly
hangc0276 · Jan 16, 2021 · 051e2b8 · 051e2b8
1 parent feb48c0
commit 051e2b8
Showing 1 changed file with 1 addition and 11 deletions.
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
@@ -15,7 +15,6 @@
 
 import static com.google.common.base.Preconditions.checkArgument;
 
-import io.netty.buffer.ByteBuf;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import org.apache.bookkeeper.mledger.AsyncCallbacks;
@@ -27,7 +26,6 @@
 import org.apache.pulsar.broker.intercept.ManagedLedgerInterceptorImpl;
 import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.client.impl.MessageIdImpl;
-import org.apache.pulsar.common.api.proto.PulsarApi;
 import org.apache.pulsar.common.protocol.Commands;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -144,16 +142,8 @@ public static PositionImpl getPositionForOffset(ManagedLedger managedLedger, Lon
         }
     }
 
-    public static PulsarApi.BrokerEntryMetadata peekBrokerEntryMetadata(ByteBuf byteBuf) {
-        final int readerIndex = byteBuf.readerIndex();
-        PulsarApi.BrokerEntryMetadata entryMetadata =
-                Commands.parseBrokerEntryMetadataIfExist(byteBuf);
-        byteBuf.readerIndex(readerIndex);
-        return entryMetadata;
-    }
-
     public static long peekOffsetFromEntry(Entry entry) {
-        return peekBrokerEntryMetadata(entry.getDataBuffer()).getIndex();
+        return Commands.peekBrokerEntryMetadataIfExist(entry.getDataBuffer()).getIndex();
     }
 
     public static long peekBaseOffsetFromEntry(Entry entry) {