devinbost edited a comment on issue #6054: URL: https://github.com/apache/pulsar/issues/6054#issuecomment-818486009
One possibility is that we're passing the wrong value (or maybe there's a race condition) when we get the number of messages when updating the permits. We use the value https://github.com/apache/pulsar/blob/master/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java#L517 which is set [here](https://github.com/apache/pulsar/blob/197bd93150a222a6ede72758eeb1d2b276dd728f/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java#L152) from [here](https://github.com/apache/pulsar/blob/197bd93150a222a6ede72758eeb1d2b276dd728f/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java#L125) which is obtained [here](https://github.com/apache/pulsar/blob/197bd93150a222a6ede72758eeb1d2b276dd728f/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java#L88). That value, `msgMetadata.getNumMessagesInBatch()`, comes from here: ``` public static MessageMetadata peekMessageMetadata(ByteBuf metadataAndPayload, String subscription, long consumerId) { try { // save the reader index and restore after parsing int readerIdx = metadataAndPayload.readerIndex(); skipBrokerEntryMetadataIfExist(metadataAndPayload); MessageMetadata metadata = Commands.parseMessageMetadata(metadataAndPayload); metadataAndPayload.readerIndex(readerIdx); return metadata; } catch (Throwable t) { log.error("[{}] [{}] Failed to parse message metadata", subscription, consumerId, t); return null; } } ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: [email protected]
