Fix the problem that batchMessageId is converted to messageIdImpl (Pu…

…lsar-8779)
cbornet · Dec 5, 2020 · 33b8d92 · 33b8d92
1 parent 56ddf5b
commit 33b8d92
Showing 1 changed file with 2 additions and 2 deletions.
diff --git a/protocol-handler/src/main/java/org/apache/pulsar/protocols/grpc/PulsarGrpcService.java b/protocol-handler/src/main/java/org/apache/pulsar/protocols/grpc/PulsarGrpcService.java
@@ -1206,15 +1206,15 @@ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
             PulsarApi.MessageMetadata metadata = parseMessageMetadata(entry.getDataBuffer());
             int batchSize = metadata.getNumMessagesInBatch();
             entry.release();
-            return batchSize;
+            return metadata.hasNumMessagesInBatch() ? batchSize : -1;
         });
 
         batchSizeFuture.whenComplete((batchSize, e) -> {
             if (e != null) {
                 responseObserver.onNext(Commands.newError(
                         requestId, ServerError.MetadataError, "Failed to get batch size for entry " + e.getMessage()));
             } else {
-                int largestBatchIndex = batchSize > 1 ? batchSize - 1 : -1;
+                int largestBatchIndex = batchSize > 0 ? batchSize - 1 : -1;
 
                 if (log.isDebugEnabled()) {
                     log.debug("[{}] [{}][{}] Get LastMessageId {} partitionIndex {}", remoteAddress,